6 Commits
Author SHA1 Message Date
LeoVasanko 3a56bfbb10 Add kanta.bootstrap logger and configurable logging setup
- Bootstrap records are now logged via kanta.bootstrap at INFO level.
- Existing databases log 'Using <path>' at DEBUG on kanta.bootstrap.
- New databases log 'Created <path>' at INFO on kanta.bootstrap.
- Renamed loggers: kanta.changes -> kanta.transaction, kanta.migrations -> kanta.migration.
- configure_logging() gains bootstrap/migration/transaction/skiproot kwargs.
- Default configure_logging() attaches a no-prefix stderr handler to kanta and stops propagation.
- With skiproot=False, child logger propagation flags are still applied but kanta itself is left untouched.
- Updated tests and docstrings.
2026-06-20 19:10:58 +00:00
LeoVasanko 55fa475a13 Log new database creation, bootstrap like a transaction. 2026-06-20 18:15:08 +00:00
LeoVasanko 753b7eba86 Add separate migration logging, logmigr callback, and log= override
- Split transaction logger (kanta.changes) and migration logger (kanta.migrations).
- Migrations.apply() now returns MigrationResult instead of logging.
- Kanta.open() emits one info summary per DB and debug transaction per migration.
- Add open(log=...) to suppress/redirect default migration logging.
- Add @kanta.logmigr callback for custom migration logging/summaries.
- Add transaction(log=...) to suppress/redirect transaction logging.
- Update tests for the new MigrationResult API and logging behaviour.
2026-06-15 23:13:03 +00:00
LeoVasanko 42789e6619 Revised migration context: don't store changerecord unless something was changed, and always snapshot if and after changes done. 2026-06-15 22:14:03 +00:00
LeoVasanko c4726e6728 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. 2026-06-15 03:09:45 +00:00
LeoVasanko 66e92739ab Refactor Migrations internals, add caching to avoid reloading per each Kanta instance. Naming changed from MigrationRegistry to Migration, module from migrate to kanta.migrations. 2026-06-15 01:12:42 +00:00
16 changed files with 952 additions and 203 deletions
+10 -1
View File
@@ -18,6 +18,7 @@ from dataclasses import dataclass
from typing import Annotated, Any, Union, get_args, get_origin
from kanta.exceptions import DatabaseError
from kanta.migrations import MigrationResult
DictPre = Annotated[dict, "pre"]
DictPost = Annotated[dict, "post"]
@@ -59,6 +60,7 @@ class InjectionContext:
error: DatabaseError | None = None
previous_state: dict | None = None
current_state: dict | None = None
migration_result: MigrationResult | None = None
@dataclass
@@ -98,6 +100,7 @@ class CallbackRegistry:
self._callbacks: dict[str, list[_CallbackRegistration]] = {
"bootstrap": [],
"fatal_error": [],
"logmigr": [],
}
self._logfmt_callbacks: list[_LogFmtFunctionSpec | _LogFmtClassSpec] = []
@@ -446,10 +449,12 @@ class CallbackRegistry:
return kind == "logfmt"
if bare is DatabaseError:
return kind == "fatal_error"
if bare is MigrationResult:
return kind == "logmigr"
if self._data_type is not None and bare is self._data_type:
return kind == "bootstrap"
if self._kanta_class is not None and bare is self._kanta_class:
return kind in {"bootstrap", "fatal_error", "logfmt"}
return kind in {"bootstrap", "fatal_error", "logfmt", "logmigr"}
return False
def _allowed_message(self, kind: str) -> str:
@@ -462,6 +467,8 @@ class CallbackRegistry:
parts.append(self._kanta_class.__name__)
if kind == "fatal_error":
parts.append("DatabaseError")
if kind == "logmigr":
parts.append("MigrationResult")
if kind == "logfmt":
parts.append("Annotated[dict, 'pre']")
parts.append("Annotated[dict, 'post']")
@@ -475,6 +482,8 @@ class CallbackRegistry:
return ctx.current_state
if bare is DatabaseError:
return ctx.error
if bare is MigrationResult:
return ctx.migration_result
if self._data_type is not None and bare is self._data_type:
return ctx.data
if self._kanta_class is not None and bare is self._kanta_class:
+40 -2
View File
@@ -1,6 +1,7 @@
"""Kanta DB main public API"""
from __future__ import annotations
import logging
from datetime import datetime
from pathlib import Path
from types import ModuleType, SimpleNamespace
@@ -144,7 +145,13 @@ class Kanta(Generic[T]):
"""
return self._impl.mtime
async def open(self, *, create: bool = True, readonly: bool = False) -> None:
async def open(
self,
*,
create: bool = True,
readonly: bool = False,
log: bool | logging.Logger = True,
) -> None:
"""Open the database file and start background persistence.
This loads existing records, applies configured migrations, and starts
@@ -156,6 +163,13 @@ class Kanta(Generic[T]):
readonly: If True, open the database read-only. No lock is acquired,
no background flush task is started, and transactions are
rejected. The file is not created if missing.
log: Controls bootstrap and migration logging. ``True`` (default)
uses the ``kanta.bootstrap`` logger for bootstrap records and
the ``kanta.migration`` logger for migration output. ``False``
suppresses the default bootstrap and migration logs. A
:class:`~logging.Logger` instance writes default output to that
logger instead. Custom ``@kanta.logmigr`` callbacks run
regardless of this setting.
Calling ``open`` more than once on the same instance is not allowed.
@@ -163,7 +177,7 @@ class Kanta(Generic[T]):
kanta.exceptions.DatabaseError: If replay or decoding fails.
kanta.exceptions.DataIntegrityError: If the instance is already open.
"""
await self._impl.open(create=create, readonly=readonly)
await self._impl.open(create=create, readonly=readonly, log=log)
async def __aenter__(self) -> Kanta[T]:
"""Enter async context manager and open the database.
@@ -239,6 +253,24 @@ class Kanta(Generic[T]):
return _register
return _register(fn)
def logmigr(self, fn=None):
"""Register a migration logging callback.
Can be used as ``@kanta.logmigr``.
The callback receives a :class:`kanta.migrations.MigrationResult` and
may be sync or async. If registered, it replaces the default migration
logger output; the application is responsible for emitting any log
messages.
"""
def _register(callback):
self._impl.add_logmigr(callback)
return callback
if fn is None:
return _register
return _register(fn)
def logfmt(self, fn=None, *, path: str | None = None):
"""Register a transaction logfmt callback.
@@ -266,6 +298,7 @@ class Kanta(Generic[T]):
*,
user: str | None = None,
mtime: bool | datetime = True,
log: bool | logging.Logger = True,
):
"""Create a transactional mutation context manager.
@@ -280,6 +313,10 @@ class Kanta(Generic[T]):
system operations that are not considered modifications. A
:class:`~datetime.datetime` value sets ``m`` to that explicit
time.
log: Controls transaction logging. ``True`` (default) uses the
``kanta.transaction`` logger. ``False`` suppresses the
transaction log. A :class:`~logging.Logger` instance writes
output to that logger instead.
Returns:
A context manager yielding the live state object for mutation.
@@ -294,4 +331,5 @@ class Kanta(Generic[T]):
action,
user=user,
mtime=mtime,
log=log,
)
+146 -24
View File
@@ -12,7 +12,8 @@ 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.logging import _USER_PATH, bootstrap_logger, log_change, migration_logger
from kanta.migrations import MigrationResult, 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 +29,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
if self.migrations is not None:
self.migrations: Migrations | None = None
if migrations is not None:
module = (
importlib.import_module(self.migrations)
if isinstance(self.migrations, str)
else self.migrations
importlib.import_module(migrations)
if isinstance(migrations, str)
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 +56,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
)
self.statedict = struct_to_dict(self.data, serializer=self.serializer)
self.version = (
self.migration_registry.dbver if self.migration_registry is not None else 0
)
self.version = self.migrations.dbver if self.migrations is not None else 0
def add_bootstrap(
self,
@@ -77,7 +76,64 @@ class KantaImpl(PersistenceMixin, Generic[T]):
"""Register one transaction logfmt callback."""
self.callback_registry.register("logfmt", callback, path=path)
async def open(self, *, create: bool = True, readonly: bool = False) -> None:
def add_logmigr(self, callback) -> None:
"""Register one migration logging callback."""
self.callback_registry.register("logmigr", callback)
async def _handle_migration_log(
self,
migration_result: MigrationResult,
previous_version: int,
log: bool | logging.Logger,
) -> None:
"""Route migration logging to callback or default logger."""
assert isinstance(migration_result, MigrationResult)
if self.callback_registry.has("logmigr"):
await self.callback_registry.invoke(
"logmigr",
InjectionContext(
kanta=self._kanta,
migration_result=migration_result,
),
)
return
if log is False:
return
migration_log = log if isinstance(log, logging.Logger) else migration_logger
changed = [m for m in migration_result.migrations if m.changed]
if not changed:
return
for info in changed:
if info.diff:
log_change(
info.name,
info.diff,
previous=info.before,
logger=migration_log,
level=logging.DEBUG,
)
descriptions = [f"{m.name} ({m.description})" for m in changed]
migration_log.info(
"Migrated %s v%s -> v%s: %s",
self.filename,
previous_version,
migration_result.version,
", ".join(descriptions),
)
async def open(
self,
*,
create: bool = True,
readonly: bool = False,
log: bool | logging.Logger = True,
) -> None:
"""Open the database: load from disk, apply migrations, start background task."""
if self.opened:
raise DataIntegrityError(
@@ -112,6 +168,9 @@ class KantaImpl(PersistenceMixin, Generic[T]):
action="open",
)
# From this point the file is open and must be closed via close().
self.opened = True
if content:
try:
rr = replay(
@@ -141,12 +200,33 @@ class KantaImpl(PersistenceMixin, Generic[T]):
cause_type=type(e).__name__,
) from e
if self.migration_registry is not None:
rr.version = self.migration_registry.apply(
migration_result = None
state_before_migrations = None
previous_version = rr.version
if self.migrations is not None:
state_before_migrations = copy.deepcopy(rr.state)
migration_result = self.migrations.apply(
rr.state, rr.version, self._kanta
)
rr.version = migration_result.version
self.statedict = copy.deepcopy(rr.state)
migrations_ran = rr.version != previous_version
migration_state_changed = (
state_before_migrations is not None
and state_before_migrations != rr.state
)
self.snapshot.ts = (
datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)
if rr.last_snapshot_mtime is not None
else None
)
self.statedict = copy.deepcopy(
state_before_migrations
if state_before_migrations is not None
else rr.state
)
self.data = restore_data_in_place(
self.data,
rr.state,
@@ -155,17 +235,35 @@ class KantaImpl(PersistenceMixin, Generic[T]):
)
self.version = rr.version
self.mtime = rr.m
if log is not False:
logger = log if isinstance(log, logging.Logger) else bootstrap_logger
logger.debug("Using %s", self.filename.resolve())
normalized = struct_to_dict(self.data, serializer=self.serializer)
if self.readonly:
self.statedict = copy.deepcopy(normalized)
else:
self.queue_change("migrate:msgspec", normalized, mtime=False)
self.snapshot.ts = (
datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)
if rr.last_snapshot_mtime is not None
else None
)
if migrations_ran and migration_state_changed:
self.queue_change(
f"migrate:v{self.version}",
rr.state,
mtime=False,
)
msgspec_record = self.queue_change(
"migrate:msgspec", normalized, mtime=False
)
if migrations_ran or msgspec_record is not None:
self.snapshot.request_force()
await self.flush()
self.snapshot.maybe_write(
self.file, self.version, self.statedict, m=self.mtime
)
if migrations_ran and migration_result is not None:
await self._handle_migration_log(
migration_result, previous_version, log
)
elif self.readonly:
self.opened = False
self.file.close()
raise DataIntegrityError(
"Cannot open empty database in read-only mode",
@@ -182,14 +280,40 @@ class KantaImpl(PersistenceMixin, Generic[T]):
self.statedict = {}
current = struct_to_dict(self.data, serializer=self.serializer)
self.queue_change(
record = self.queue_change(
self.bootstrap_action,
current,
user=self.bootstrap_user,
mtime=self.bootstrap_mtime,
force=True,
)
if record is not None and log is not False:
logger = log if isinstance(log, logging.Logger) else bootstrap_logger
logger.info("Created %s", self.filename.resolve())
logfmt = self.callback_registry.build_logfmt(
InjectionContext(
previous_state={},
current_state=current,
kanta=self._kanta,
)
)
formatted_user = self.bootstrap_user
if formatted_user is not None and logfmt is not None:
resolved = logfmt(formatted_user, _USER_PATH)
if resolved is not None:
formatted_user = resolved
log_change(
self.bootstrap_action,
record.diff,
formatted_user,
previous={},
logfmt=logfmt,
logger=logger,
level=logging.INFO,
)
except Exception:
self.opened = False
self.file.close()
try:
await asyncio.to_thread(self.filename.unlink, missing_ok=True)
@@ -197,8 +321,6 @@ class KantaImpl(PersistenceMixin, Generic[T]):
pass
raise
self.opened = True
if not self.readonly:
self.background_task = asyncio.create_task(self._background_loop())
+55 -13
View File
@@ -1,7 +1,8 @@
"""Database change logging with pretty-printed diffs.
Provides a logger for JSONL database changes that formats diffs
in a human-readable path.notation style with color coding.
Provides loggers for JSONL database changes, bootstrap events, and
migrations. Diff output is formatted in a human-readable path notation
style with color coding.
"""
import logging
@@ -10,7 +11,9 @@ import sys
from collections.abc import Callable
from typing import Any
logger = logging.getLogger("kanta.changes")
transaction_logger = logging.getLogger("kanta.transaction")
bootstrap_logger = logging.getLogger("kanta.bootstrap")
migration_logger = logging.getLogger("kanta.migration")
# Pattern to match control characters and bidirectional overrides
_UNSAFE_CHARS = re.compile(
@@ -274,6 +277,9 @@ def log_change(
user: str | None = None,
previous: dict | None = None,
logfmt: Callable[[Any, str], str | None] | None = None,
*,
logger: logging.Logger = transaction_logger,
level: int = logging.INFO,
) -> None:
"""Log a database change with pretty-printed diff.
@@ -283,27 +289,63 @@ def log_change(
user: Optional already-formatted user name to show in the header.
previous: The previous state dict (for determining add vs update).
logfmt: Optional formatter callable ``(value, path) -> str | None``.
logger: Logger to write to. Defaults to the ``kanta.transaction`` logger.
level: Log level to use. Defaults to ``logging.INFO``.
"""
header = format_action_header(action, user)
diff_lines = format_diff(diff, previous, logfmt)
if not diff_lines:
logger.info(header)
logger.log(level, header)
return
if len(diff_lines) == 1:
logger.info(f"{header}{diff_lines[0]}")
logger.log(level, f"{header}{diff_lines[0]}")
else:
logger.info(header)
logger.log(level, header)
for line in diff_lines:
logger.info(line)
logger.log(level, line)
def configure_logging() -> None:
"""Configure the database logger to output to stderr without prefix."""
if not logger.handlers:
def configure_logging(
*,
skiproot: bool = True,
bootstrap: bool = True,
migration: bool = True,
transaction: bool = True,
) -> None:
"""Configure Kanta's default logging output.
Args:
skiproot: If ``True`` (default), attach a no-prefix stderr handler to
the ``kanta`` logger and set ``kanta.propagate = False`` so Kanta
output is rendered directly without propagating to the root logger.
If ``False``, the child logger enable flags are still applied, but
no handler is added and ``kanta`` propagation is left untouched so
the application's root logger handles Kanta output.
bootstrap: Whether bootstrap logs are enabled.
migration: Whether migration logs are enabled.
transaction: Whether transaction logs are enabled.
This helper is not called automatically; applications that want Kanta's
default output can call it, but most applications will configure logging
themselves.
"""
for name, enabled in (
("kanta.bootstrap", bootstrap),
("kanta.migration", migration),
("kanta.transaction", transaction),
):
logging.getLogger(name).propagate = enabled
if not skiproot:
return
target = logging.getLogger("kanta")
target.propagate = False
if not target.handlers:
handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(handler)
logger.setLevel(logging.INFO)
logger.propagate = False
target.addHandler(handler)
target.setLevel(logging.INFO)
-122
View File
@@ -1,122 +0,0 @@
"""Database schema migration framework.
Migrations are numbered functions discovered automatically via a decorator
or by prefix. Each runs exactly once based on the current version.
"""
from __future__ import annotations
import importlib
import inspect
import logging
from types import ModuleType
from typing import Any
_logger = logging.getLogger(__name__)
class MigrationRegistry:
"""Registry of schema migration functions.
Usage::
registry = MigrationRegistry()
@registry.register
def migrate_v1(d: dict, kanta) -> None:
d.setdefault("version", 1)
kanta.ctx.note = "migrated"
@registry.register
def migrate_v2(d: dict) -> None:
d.setdefault("version", 2)
new_version = registry.apply(state, current_version=0, kanta=kanta)
Or load from a module::
registry = MigrationRegistry.from_module("myapp.migrations")
new_version = registry.apply(state, current_version=0, kanta=kanta)
"""
def __init__(self) -> None:
self._migrations: dict[int, Any] = {}
@staticmethod
def _migration_version(fn: Any) -> int:
name = getattr(fn, "__name__", "")
if not name.startswith("migrate_v"):
raise ValueError(f"Invalid migration function name: {name!r}")
suffix = name.removeprefix("migrate_v")
if not suffix.isdigit() or int(suffix) <= 0:
raise ValueError(f"Invalid migration version in function name: {name!r}")
return int(suffix)
def register(self, fn):
"""Decorator to register a migration function."""
version = self._migration_version(fn)
self._migrations[version] = fn
return fn
@classmethod
def from_module(cls, module: str | ModuleType) -> MigrationRegistry:
"""Create a registry by scanning a module for ``migrate_vN`` functions.
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
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
return reg
@property
def dbver(self) -> int:
"""Current schema version (= highest discovered migration, or 0)."""
return max(self._migrations.keys(), default=0)
@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."""
try:
inspect.signature(fn).bind(data_dict, kanta)
except TypeError:
fn(data_dict)
else:
fn(data_dict, kanta)
def apply(
self,
data_dict: dict[str, Any],
current_version: int,
kanta: Any,
*,
silent: bool = False,
) -> int:
"""Apply pending migrations to *data_dict* in place.
Returns the new version after all migrations.
"""
while current_version < self.dbver:
next_version = current_version + 1
fn = self._migrations.get(next_version)
if fn is None:
raise ValueError(
f"Missing migration step migrate_v{next_version} "
f"(highest discovered is v{self.dbver})"
)
self._call_migration(fn, data_dict, kanta)
current_version = next_version
if not silent:
desc = (fn.__doc__ or fn.__name__).split("\n")[0].rstrip(".")
_logger.info("Applied migration %s: %s", fn.__name__, desc)
return current_version
+184
View File
@@ -0,0 +1,184 @@
"""Database schema migration framework.
Migrations are numbered functions discovered automatically via a decorator
or by prefix. Each runs exactly once based on the current version.
"""
from __future__ import annotations
import copy
import importlib
import inspect
from dataclasses import dataclass
from types import ModuleType
from typing import Any
from kanta.diff import compute_diff
from kanta.exceptions import DatabaseError
# 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] = {}
@dataclass
class MigrationInfo:
"""Information about a single migration that ran."""
name: str
description: str
version: int
changed: bool
diff: dict | None = None
before: dict | None = None
@dataclass
class MigrationResult:
"""Result of applying migrations."""
version: int
migrations: list[MigrationInfo]
class Migrations:
"""Registry of schema migration functions.
Usage::
migrations = Migrations()
@migrations.register
def migrate_v1(d: dict, kanta) -> None:
d.setdefault("version", 1)
kanta.ctx.note = "migrated"
@migrations.register
def migrate_v2(d: dict) -> None:
d.setdefault("version", 2)
result = migrations.apply(state, current_version=0, kanta=kanta)
new_version = result.version
Or load from a module::
migrations = Migrations.from_module("myapp.migrations")
result = migrations.apply(state, current_version=0, kanta=kanta)
"""
def __init__(self) -> None:
self._migrations: dict[int, Any] = {}
@staticmethod
def _migration_version(fn: Any) -> int:
name = getattr(fn, "__name__", "")
if not name.startswith("migrate_v"):
raise ValueError(f"Invalid migration function name: {name!r}")
suffix = name.removeprefix("migrate_v")
if not suffix.isdigit() or int(suffix) <= 0:
raise ValueError(f"Invalid migration version in function name: {name!r}")
return int(suffix)
def register(self, fn):
"""Decorator to register a migration function."""
version = self._migration_version(fn)
self._migrations[version] = fn
return fn
@classmethod
def from_module(cls, module: str | ModuleType) -> Migrations:
"""Create or retrieve a cached registry by scanning a module.
Args:
module: A module name (string) or an imported module object.
"""
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
def dbver(self) -> int:
"""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."""
try:
inspect.signature(fn).bind(data_dict, kanta)
except TypeError:
fn(data_dict)
else:
fn(data_dict, kanta)
def apply(
self,
data_dict: dict[str, Any],
current_version: int,
kanta: Any,
) -> MigrationResult:
"""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 a :class:`MigrationResult` describing the new version and every
migration that ran.
"""
if current_version > self.dbver:
raise DatabaseError(
f"Database version v{current_version} is newer than the "
f"highest supported version v{self.dbver}"
)
if current_version < self.minver:
raise DatabaseError(
f"Database version v{current_version} is older than the "
f"minimum supported version v{self.minver}"
)
migrations: list[MigrationInfo] = []
for version in sorted(self._migrations.keys()):
if version <= current_version:
continue
fn = self._migrations[version]
before = copy.deepcopy(data_dict)
self._call_migration(fn, data_dict, kanta)
current_version = version
changed = before != data_dict
diff = compute_diff(before, data_dict) if changed else None
desc = (fn.__doc__ or f"v{version}").split("\n")[0].rstrip(".")
migrations.append(
MigrationInfo(
name=fn.__name__,
description=desc,
version=version,
changed=changed,
diff=diff,
before=before,
)
)
return MigrationResult(version=current_version, migrations=migrations)
+4 -2
View File
@@ -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:
return None
if not diff:
if not force:
return None
diff = {}
record = ChangeRecord(
ts=now,
+8 -7
View File
@@ -41,15 +41,16 @@ class SnapshotState:
self, file, version: int, state: dict, m: datetime | None = None
) -> None:
"""Write snapshot when thresholds/time policy allows it."""
if self.changes < self._min_diffs:
return
force = self._force_pending
now = datetime.now(UTC)
if not force and now.weekday() != 6: # 6 = Sunday
return
sunday_midnight = now.replace(hour=0, minute=0, second=0, microsecond=0)
if not force and self.ts is not None and self.ts >= sunday_midnight:
return
if not force:
if self.changes < self._min_diffs:
return
if now.weekday() != 6: # 6 = Sunday
return
sunday_midnight = now.replace(hour=0, minute=0, second=0, microsecond=0)
if self.ts is not None and self.ts >= sunday_midnight:
return
if not file.is_open:
return
try:
+1 -1
View File
@@ -23,7 +23,7 @@ class ChangeRecord(msgspec.Struct, omit_defaults=True, kw_only=True):
v: int = 0
u: str | None = None
m: datetime | None = None
diff: dict
diff: dict = {}
class Snapshot(msgspec.Struct, omit_defaults=True):
+12 -2
View File
@@ -9,7 +9,7 @@ from datetime import datetime
from kanta.diff import compute_diff
from kanta.exceptions import DataIntegrityError
from kanta.callbacks import InjectionContext
from kanta.logging import _USER_PATH, log_change
from kanta.logging import _USER_PATH, log_change, transaction_logger
from kanta.serialization import restore_data_in_place, struct_to_dict
_logger = logging.getLogger(__name__)
@@ -22,6 +22,7 @@ def transaction(
*,
user: str | None = None,
mtime: bool | datetime = True,
log: bool | logging.Logger = True,
):
"""Wrap writes in a transaction and yield the live db object."""
if impl.readonly:
@@ -80,7 +81,16 @@ def transaction(
resolved = logfmt(user, _USER_PATH)
if resolved is not None:
formatted_user = resolved
log_change(action, record.diff, formatted_user, previous, logfmt)
if log is not False:
logger = log if isinstance(log, logging.Logger) else transaction_logger
log_change(
action,
record.diff,
formatted_user,
previous,
logfmt,
logger=logger,
)
except Exception:
_logger.warning("Transaction '%s' failed, rolling back changes", action)
if impl.transaction_snapshot is not None:
+12 -1
View File
@@ -7,7 +7,7 @@ from uuid import UUID
import msgspec
from kanta.kanta import Kanta
from kanta.structs import ChangeRecord
from kanta.structs import ChangeRecord, Snapshot
class User(msgspec.Struct):
@@ -89,6 +89,17 @@ def make_migrations_module(name: str, fn_name: str, fn):
return mod
def read_last_snapshot(path: Path, format_config) -> Snapshot | None:
_, serializer_cls = format_config
serializer = serializer_cls()
framer = serializer.framer_cls()
data = path.read_bytes()
payload, _, _ = framer.scan_last_snapshot(data)
if payload is None:
return None
return serializer.decode(payload, type=Snapshot)
def fixed_change(action: str, diff: dict, *, version: int = 0) -> ChangeRecord:
return ChangeRecord(
ts=datetime(2026, 1, 1, tzinfo=UTC), a=action, v=version, diff=diff
+9 -7
View File
@@ -148,7 +148,7 @@ async def test_bootstrap_injects_kanta(tmp_path, format_config):
async def test_logfmt_injects_states(tmp_path, format_config, caplog):
import logging
caplog.set_level(logging.INFO, logger="kanta.changes")
caplog.set_level(logging.INFO, logger="kanta.transaction")
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
@@ -170,13 +170,15 @@ async def test_logfmt_injects_states(tmp_path, format_config, caplog):
async def test_logfmt_class_injection(tmp_path, format_config, caplog):
import logging
caplog.set_level(logging.INFO, logger="kanta.changes")
caplog.set_level(logging.INFO, logger="kanta.transaction")
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
@kanta.logfmt
class UserLogFmt(LogFmt):
def resolve(self, value: str, path: str) -> str | None:
if not isinstance(value, str):
return None
return self.current_state.get("users", {}).get(value, {}).get("name")
await kanta.open()
@@ -193,7 +195,7 @@ async def test_logfmt_class_injection(tmp_path, format_config, caplog):
async def test_multiple_logfmt_chain(tmp_path, format_config, caplog):
import logging
caplog.set_level(logging.INFO, logger="kanta.changes")
caplog.set_level(logging.INFO, logger="kanta.transaction")
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
@@ -221,7 +223,7 @@ async def test_multiple_logfmt_chain(tmp_path, format_config, caplog):
async def test_logfmt_path_context(tmp_path, format_config, caplog):
import logging
caplog.set_level(logging.INFO, logger="kanta.changes")
caplog.set_level(logging.INFO, logger="kanta.transaction")
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
@@ -245,7 +247,7 @@ async def test_logfmt_path_context(tmp_path, format_config, caplog):
async def test_logfmt_decorator_path_filters_calls(tmp_path, format_config, caplog):
import logging
caplog.set_level(logging.INFO, logger="kanta.changes")
caplog.set_level(logging.INFO, logger="kanta.transaction")
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
@@ -271,7 +273,7 @@ async def test_logfmt_decorator_path_filters_calls(tmp_path, format_config, capl
async def test_logfmt_user_path_replaces_user_display(tmp_path, format_config, caplog):
import logging
caplog.set_level(logging.INFO, logger="kanta.changes")
caplog.set_level(logging.INFO, logger="kanta.transaction")
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
@@ -293,7 +295,7 @@ async def test_logfmt_user_path_replaces_user_display(tmp_path, format_config, c
async def test_logfmt_non_string_value(tmp_path, format_config, caplog):
import logging
caplog.set_level(logging.INFO, logger="kanta.changes")
caplog.set_level(logging.INFO, logger="kanta.transaction")
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
+285
View File
@@ -1,4 +1,5 @@
import asyncio
import logging
import sys
from datetime import UTC, datetime
from uuid import uuid4
@@ -6,6 +7,7 @@ from uuid import uuid4
import pytest
from kanta.exceptions import DatabaseError, DataIntegrityError, FileLockError
from kanta.migrations import MigrationResult
from kanta.serialization import struct_to_dict
from .support import (
@@ -17,7 +19,9 @@ from .support import (
change_actions,
fixed_change,
make_kanta,
make_migrations_module,
read_changes,
read_last_snapshot,
seed_single_change,
)
@@ -69,6 +73,25 @@ async def test_new_file_persists_initial_state_for_roundtrip(tmp_path, format_co
await kanta2.close()
@pytest.mark.asyncio
async def test_reopen_without_changes_does_not_force_snapshot(tmp_path, format_config):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data(counter=5), format_config)
await kanta.open()
await kanta.close()
# No snapshot should exist after the initial bootstrap and close.
assert read_last_snapshot(path, format_config) is None
kanta2 = make_kanta(path, Data, format_config)
await kanta2.open()
assert kanta2.data.counter == 5
await kanta2.close()
# Re-opening without migrations or normalization changes must not force one.
assert read_last_snapshot(path, format_config) is None
@pytest.mark.asyncio
async def test_open_overwrites_caller_owned_root_data(tmp_path, format_config):
path = tmp_path / "test.db"
@@ -473,6 +496,268 @@ 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_writes_snapshot_and_is_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.close()
# Empty migrations must not produce empty change records.
records = read_changes(path, format_config)
migration_records = [r for r in records if r.a.startswith("migrate")]
assert not migration_records
# The version bump is persisted via a snapshot instead.
snap = read_last_snapshot(path, format_config)
assert snap is not None
assert snap.v == 1
assert snap.state == {"counter": 0, "users": {}}
kanta2 = make_kanta(path, Data, format_config, migrations=mod)
await kanta2.open()
assert kanta2.version == 1
await kanta2.close()
# Re-opening must not create additional migration records or snapshots.
records2 = read_changes(path, format_config)
assert not [r for r in records2 if r.a.startswith("migrate")]
finally:
sys.modules.pop("empty_migration_mod", None)
@pytest.mark.asyncio
async def test_migration_with_changes_records_diff_and_snapshot(
tmp_path, format_config
):
path = tmp_path / "test.db"
seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
mod = type(sys)("test_migrations_changes")
def migrate_v1(d, kanta):
d["counter"] = 2
mod.__dict__["migrate_v1"] = migrate_v1
kanta = make_kanta(path, Data, format_config, migrations=mod)
await kanta.open()
assert kanta.version == 1
assert kanta.data.counter == 2
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) == 2
assert migration_records[0].a == "migrate:v1"
assert migration_records[0].v == 1
assert migration_records[0].diff == {"counter": 2}
assert migration_records[1].a == "migrate:msgspec"
assert migration_records[1].v == 1
assert migration_records[1].diff == {"users": {}}
snap = read_last_snapshot(path, format_config)
assert snap is not None
assert snap.v == 1
assert snap.state == {"counter": 2, "users": {}}
@pytest.mark.asyncio
async def test_migration_summary_log_includes_filename(tmp_path, format_config, caplog):
path = tmp_path / "test.db"
seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
mod = type(sys)("test_migrations_log")
def migrate_v1(d, kanta):
"""Bump counter."""
d["counter"] = 2
mod.__dict__["migrate_v1"] = migrate_v1
with caplog.at_level(logging.INFO, logger="kanta.migration"):
kanta = make_kanta(path, Data, format_config, migrations=mod)
await kanta.open()
assert kanta.version == 1
await kanta.close()
info_messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
assert len(info_messages) == 1
assert str(path) in info_messages[0]
assert "v0 -> v1" in info_messages[0]
assert "migrate_v1 (Bump counter)" in info_messages[0]
@pytest.mark.asyncio
async def test_open_log_false_suppresses_migration_log(tmp_path, format_config, caplog):
path = tmp_path / "test.db"
seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
mod = type(sys)("test_migrations_silent")
def migrate_v1(d, kanta):
d["counter"] = 2
mod.__dict__["migrate_v1"] = migrate_v1
with caplog.at_level(logging.INFO, logger="kanta.migration"):
kanta = make_kanta(path, Data, format_config, migrations=mod)
await kanta.open(log=False)
await kanta.close()
info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
assert not info_messages
@pytest.mark.asyncio
async def test_open_log_true_logs_bootstrap(tmp_path, format_config, caplog):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
with caplog.at_level(logging.INFO, logger="kanta.bootstrap"):
await kanta.open()
await kanta.close()
info_messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
assert len(info_messages) >= 2
assert "Created" in info_messages[0]
assert "bootstrap" in info_messages[1]
@pytest.mark.asyncio
async def test_open_log_false_suppresses_bootstrap_log(tmp_path, format_config, caplog):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
with caplog.at_level(logging.INFO, logger="kanta.bootstrap"):
await kanta.open(log=False)
await kanta.close()
info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
assert not info_messages
@pytest.mark.asyncio
async def test_open_log_custom_logger_logs_bootstrap(tmp_path, format_config, caplog):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
custom_logger = logging.getLogger("custom.bootstrap")
custom_logger.setLevel(logging.INFO)
with caplog.at_level(logging.INFO, logger="custom.bootstrap"):
await kanta.open(log=custom_logger)
await kanta.close()
info_messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
assert len(info_messages) >= 2
assert "Created" in info_messages[0]
assert "bootstrap" in info_messages[1]
@pytest.mark.asyncio
async def test_open_existing_database_logs_using_on_debug(
tmp_path, format_config, caplog
):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
await kanta.open()
await kanta.close()
kanta2 = make_kanta(path, Data, format_config)
with caplog.at_level(logging.DEBUG, logger="kanta.bootstrap"):
await kanta2.open()
await kanta2.close()
debug_messages = [r.message for r in caplog.records if r.levelno == logging.DEBUG]
assert any("Using" in m and str(path.resolve()) in m for m in debug_messages)
@pytest.mark.asyncio
async def test_logmigr_callback_replaces_default_logging(
tmp_path, format_config, caplog
):
path = tmp_path / "test.db"
seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
mod = type(sys)("test_migrations_callback")
def migrate_v1(d, kanta):
"""Bump counter."""
d["counter"] = 2
mod.__dict__["migrate_v1"] = migrate_v1
summaries = []
kanta = make_kanta(path, Data, format_config, migrations=mod)
@kanta.logmigr
def collect(summary: MigrationResult):
summaries.append(summary)
with caplog.at_level(logging.INFO, logger="kanta.migration"):
await kanta.open()
await kanta.close()
assert len(summaries) == 1
assert summaries[0].version == 1
assert summaries[0].migrations[0].name == "migrate_v1"
info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
assert not info_messages
@pytest.mark.asyncio
async def test_transaction_log_false_suppresses_log(tmp_path, format_config, caplog):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
await kanta.open()
with caplog.at_level(logging.INFO, logger="kanta.transaction"):
with kanta.transaction(action="inc", log=False) as data:
data.counter = 1
await kanta.close()
info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
assert not info_messages
@pytest.mark.asyncio
async def test_transaction_log_custom_logger(tmp_path, format_config, caplog):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
await kanta.open()
custom_logger = logging.getLogger("custom.transaction")
custom_logger.setLevel(logging.INFO)
with caplog.at_level(logging.INFO, logger="custom.transaction"):
with kanta.transaction(action="inc", log=custom_logger) as data:
data.counter = 1
await kanta.close()
info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
assert len(info_messages) >= 1
assert "inc" in info_messages[0].message
@pytest.mark.asyncio
async def test_open_locked_file_raises_filelock_error(tmp_path, format_config):
path = tmp_path / "test.db"
+36 -5
View File
@@ -1,16 +1,47 @@
import logging
from kanta.logging import configure_logging, log_change
from kanta.logging import logger
import pytest
from kanta.logging import configure_logging, log_change, transaction_logger
def test_configure_logging():
@pytest.fixture(autouse=True)
def _reset_kanta_loggers():
yield
for name in ("kanta", "kanta.transaction", "kanta.bootstrap", "kanta.migration"):
logger = logging.getLogger(name)
logger.setLevel(logging.NOTSET)
logger.propagate = True
logger.handlers.clear()
def test_configure_logging_defaults():
kanta_logger = logging.getLogger("kanta")
configure_logging()
assert logger.level == logging.INFO
assert kanta_logger.level == logging.INFO
assert not kanta_logger.propagate
assert kanta_logger.handlers
def test_configure_logging_disables_specific_loggers():
configure_logging(bootstrap=False, migration=False, transaction=False)
assert not logging.getLogger("kanta.bootstrap").propagate
assert not logging.getLogger("kanta.migration").propagate
assert not logging.getLogger("kanta.transaction").propagate
def test_configure_logging_skiproot_false_leaves_kanta_propagation():
kanta_logger = logging.getLogger("kanta")
kanta_logger.handlers.clear()
configure_logging(bootstrap=False, skiproot=False)
assert kanta_logger.propagate
assert not kanta_logger.handlers
assert not logging.getLogger("kanta.bootstrap").propagate
def test_log_change_no_diff(capsys):
logger.handlers.clear()
kanta_logger = logging.getLogger("kanta")
kanta_logger.handlers.clear()
configure_logging()
log_change("test", {})
captured = capsys.readouterr()
+133 -16
View File
@@ -1,6 +1,9 @@
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 +12,7 @@ class _DummyKanta:
def test_register_and_apply():
reg = MigrationRegistry()
reg = Migrations()
kanta = _DummyKanta()
@reg.register
@@ -21,13 +24,13 @@ def test_register_and_apply():
d["version"] = 2
state = {}
new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
assert new_ver == 2
result = reg.apply(state, current_version=0, kanta=kanta)
assert result.version == 2
assert state["version"] == 2
def test_no_migrations_needed():
reg = MigrationRegistry()
reg = Migrations()
kanta = _DummyKanta()
@reg.register
@@ -35,8 +38,8 @@ def test_no_migrations_needed():
d["x"] = 1
state = {"x": 1}
new_ver = reg.apply(state, current_version=1, kanta=kanta, silent=True)
assert new_ver == 1
result = reg.apply(state, current_version=1, kanta=kanta)
assert result.version == 1
def test_from_module():
@@ -52,17 +55,17 @@ 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 = {}
new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
assert new_ver == 2
result = reg.apply(state, current_version=0, kanta=kanta)
assert result.version == 2
assert state["v"] == 2
def test_migrations_can_use_kanta_ctx():
reg = MigrationRegistry()
reg = Migrations()
kanta = _DummyKanta()
@reg.register
@@ -71,14 +74,14 @@ def test_migrations_can_use_kanta_ctx():
d["source"] = kanta.ctx.source
state = {}
new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
assert new_ver == 1
result = reg.apply(state, current_version=0, kanta=kanta)
assert result.version == 1
assert state["source"] == "migration"
assert kanta.ctx.source == "migration"
def test_migration_can_omit_kanta_argument():
reg = MigrationRegistry()
reg = Migrations()
kanta = _DummyKanta()
@reg.register
@@ -86,6 +89,120 @@ def test_migration_can_omit_kanta_argument():
d["x"] = 1
state = {}
new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
assert new_ver == 1
result = reg.apply(state, current_version=0, kanta=kanta)
assert result.version == 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)
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)
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}
result = reg.apply(state, current_version=1, kanta=kanta)
assert result.version == 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}
result = reg.apply(state, current_version=2, kanta=kanta)
assert result.version == 3
assert state["x"] == 3
def test_apply_returns_change_information():
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
result = reg.apply({}, current_version=0, kanta=kanta)
assert result.version == 3
assert len(result.migrations) == 3
assert result.migrations[0].name == "migrate_v1"
assert result.migrations[0].description == "Set x"
assert result.migrations[0].changed is True
assert result.migrations[0].diff == {"$replace": {"x": 1}}
assert result.migrations[1].name == "migrate_v2"
assert result.migrations[1].description == "No-op"
assert result.migrations[1].changed is False
assert result.migrations[1].diff is None
assert result.migrations[2].name == "migrate_v3"
assert result.migrations[2].description == "Set y"
assert result.migrations[2].changed is True
assert result.migrations[2].diff == {"y": 3}
def test_description_defaults_to_version_when_no_docstring():
reg = Migrations()
kanta = _DummyKanta()
@reg.register
def migrate_v1(d):
d["x"] = 1
result = reg.apply({}, current_version=0, kanta=kanta)
assert result.migrations[0].description == "v1"
+17
View File
@@ -32,3 +32,20 @@ def test_force_writes():
f = FakeFile()
ss.maybe_write(f, 1, {"x": 1})
assert len(f.written) == 1
def test_force_bypasses_min_diffs():
class FakeFile:
def __init__(self):
self.written = []
self.is_open = True
def write(self, data: bytes):
self.written.append(data)
ss = SnapshotState(min_diffs=100)
ss.record_changes(5)
ss.request_force()
f = FakeFile()
ss.maybe_write(f, 1, {"x": 1})
assert len(f.written) == 1