7 Commits
Author SHA1 Message Date
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
LeoVasanko 4dc2f0648e Fix creation of new database, ensuring that a bootstrap record of the initial state is always written. 2026-06-15 00:27:49 +00:00
LeoVasanko bec4635460 Add kanta.ctx SimpleNamespace for user variables. This is passed to migration functions if they take a second argument. Remove the old MigrationCtx system. 2026-06-14 19:54:23 +00:00
LeoVasanko c04a245366 Add read only mode that doesn't use locking. Useful for inspection while the database is in use or when no changes are intended (except in RAM). 2026-06-14 19:35:06 +00:00
21 changed files with 1237 additions and 266 deletions
+21 -5
View File
@@ -53,10 +53,13 @@ asyncio.run(main())
## Bootstrap and Open Modes
Kanta supports open-time bootstrap callbacks for initializing a brand-new
database before `open()` returns.
When `open()` creates a brand-new database, it always writes a single bootstrap
change record from the initial data object you passed to `Kanta(...)`. The
simplest bootstrap is therefore the object itself — no extra code is required.
Register bootstrap handlers with a decorator:
Bootstrap handlers are optional. Use them only when you need to modify the
initial state at creation time, for example to seed defaults or perform
expensive/external setup that should happen exactly once:
```python
kanta = Kanta("data.kantadb", Data())
@@ -76,9 +79,10 @@ async def bootstrap_async(data) -> None:
data.counter = 1
```
When multiple bootstrap handlers are registered:
Whether or not handlers are registered, exactly one bootstrap change record is
written when a new database is created. The record contains the initial object,
or the state after all bootstrap handlers have run. When handlers are present:
- they run in registration order,
- exactly one bootstrap change record is queued,
- bootstrap metadata (`action`, `user`, `mtime`) is taken from the last
registration.
@@ -94,6 +98,18 @@ await kanta.open(create=False)
With `create=False`, open fails if the database file does not exist or is
empty.
Read-only mode opens an existing database without locking it or starting the
background flush task. This is useful for readers that must not block the
writer or modify the file:
```python
await kanta.open(readonly=True)
```
In read-only mode, records are replayed and migrations are applied in memory,
but transactions and explicit flushes are rejected and the file is never
created if missing.
## Fatal Error Handlers
Fatal background write errors can be observed with a decorator:
+13 -1
View File
@@ -118,6 +118,12 @@ reloads, while system operations such as migrations leave it unchanged.
- `await kanta.open()` (default) creates the database file if missing.
- `await kanta.open(create=False)` fails when the file is missing or empty.
- `await kanta.open(readonly=True)` opens an existing database read-only.
- The file is opened without acquiring a lock and without a background flush
task.
- Existing records are replayed and migrations are still applied in memory.
- Transactions and explicit flushes are rejected.
- The file is never created if missing.
### Callbacks
@@ -127,7 +133,11 @@ when they have a default value.
#### Bootstrap Callbacks
- Bootstrap callbacks run during `open()` when the database is empty.
- When `open()` creates a new database, it always writes a single bootstrap
`ChangeRecord`.
- The simplest bootstrap is the initial data object passed to `Kanta(...)`;
bootstrap callbacks are optional and only needed when you want to modify or
enrich that object at creation time.
- Register callbacks via:
- `@kanta.bootstrap`
- `@kanta.bootstrap(action=..., user=..., mtime=...)`
@@ -140,6 +150,8 @@ when they have a default value.
- exactly one bootstrap `ChangeRecord` is queued,
- bootstrap metadata (`action`, `user`, `mtime`) is taken from the last
callback registration.
- If no bootstrap callbacks are registered, the bootstrap record still uses
`action="bootstrap"` and contains the initial data object.
- If any bootstrap callback raises, Kanta closes and removes the database file,
then re-raises the exception.
+28 -26
View File
@@ -12,11 +12,13 @@ and receive the value plus an optional ``path`` string. They return
from __future__ import annotations
import inspect
import types
from collections.abc import Callable
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"]
@@ -58,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
@@ -97,6 +100,7 @@ class CallbackRegistry:
self._callbacks: dict[str, list[_CallbackRegistration]] = {
"bootstrap": [],
"fatal_error": [],
"logmigr": [],
}
self._logfmt_callbacks: list[_LogFmtFunctionSpec | _LogFmtClassSpec] = []
@@ -317,17 +321,13 @@ class CallbackRegistry:
f"Allowed: str path, {self._allowed_message('logfmt')}"
)
if sig.return_annotation is inspect.Signature.empty:
raise TypeError(
f"logfmt callback {callback.__name__} must annotate its "
f"return type as str | None"
)
return_ann = self._resolve_raw_annotation(sig.return_annotation, callback)
if not self._is_optional_str(return_ann):
raise TypeError(
f"logfmt callback {callback.__name__} must return str | None, "
f"got {return_ann!r}"
)
if sig.return_annotation is not inspect.Signature.empty:
return_ann = self._resolve_raw_annotation(sig.return_annotation, callback)
if not self._is_optional_str(return_ann):
raise TypeError(
f"logfmt callback {callback.__name__} must return str | None, "
f"got {return_ann!r}"
)
return _LogFmtFunctionSpec(
callback=callback,
@@ -416,19 +416,15 @@ class CallbackRegistry:
f"logfmt class {cls.__name__}.resolve must accept a 'path: str' parameter"
)
if resolve_sig.return_annotation is inspect.Signature.empty:
raise TypeError(
f"logfmt class {cls.__name__}.resolve must annotate its "
f"return type as str | None"
)
return_ann = self._resolve_raw_annotation(
resolve_sig.return_annotation, resolve
)
if not self._is_optional_str(return_ann):
raise TypeError(
f"logfmt class {cls.__name__}.resolve must return str | None, "
f"got {return_ann!r}"
if resolve_sig.return_annotation is not inspect.Signature.empty:
return_ann = self._resolve_raw_annotation(
resolve_sig.return_annotation, resolve
)
if not self._is_optional_str(return_ann):
raise TypeError(
f"logfmt class {cls.__name__}.resolve must return str | None, "
f"got {return_ann!r}"
)
return _LogFmtClassSpec(cls=cls, inject_params=inject_params, path=path)
@@ -453,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:
@@ -469,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']")
@@ -482,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:
@@ -516,7 +518,7 @@ class CallbackRegistry:
@staticmethod
def _unwrap_optional(ann: Any) -> Any:
origin = get_origin(ann)
if origin is not Union:
if origin not in (Union, types.UnionType):
return ann
args = [arg for arg in get_args(ann) if arg is not type(None)]
return args[0] if len(args) == 1 else ann
@@ -524,7 +526,7 @@ class CallbackRegistry:
@staticmethod
def _is_optional_str(ann: Any) -> bool:
origin = get_origin(ann)
if origin is not Union:
if origin not in (Union, types.UnionType):
return ann is str
args = get_args(ann)
return type(None) in args and any(arg is str for arg in args)
+42 -26
View File
@@ -34,6 +34,7 @@ if sys.platform == "win32":
_GENERIC_READ = 0x80000000
_GENERIC_WRITE = 0x40000000
_FILE_SHARE_READ = 0x00000001
_FILE_SHARE_WRITE = 0x00000002
_OPEN_EXISTING = 3
_OPEN_ALWAYS = 4
_FILE_ATTRIBUTE_NORMAL = 0x80
@@ -91,15 +92,16 @@ else:
class LockedFile:
"""A file opened with an exclusive write lock.
"""A file opened for read+write with an optional exclusive lock.
Usage::
f = LockedFile()
f.open(path) # open + lock (read+write)
content = f.read() # read entire content
f.write(data) # append data (seeks to end first)
f.close() # release lock + close fd
f.open(path) # open + lock (read+write)
f.open(path, readonly=True) # open read-only without locking
content = f.read() # read entire content
f.write(data) # append data (seeks to end first)
f.close() # release lock + close fd
Unix: fcntl.flock (advisory) — read-only callers that don't flock are unaffected.
Windows: CreateFileW with FILE_SHARE_READ — OS blocks other writers.
@@ -108,12 +110,13 @@ class LockedFile:
def __init__(self) -> None:
self._fd: int | None = None # Unix fd or Windows HANDLE
def open(self, path: Path, *, create: bool = False) -> None:
"""Open *path* for read+write with an exclusive lock.
def open(self, path: Path, *, create: bool = False, readonly: bool = False) -> None:
"""Open *path* and optionally acquire an exclusive lock.
Args:
path: File to open and lock.
create: If True, create the file if it doesn't exist (bootstrap).
readonly: If True, open read-only without acquiring a lock.
Raises:
FileLockError: If the file is locked by another process or not found.
@@ -122,16 +125,18 @@ class LockedFile:
return # Already open (idempotent)
if sys.platform == "win32":
self._open_win32(path, create)
self._open_win32(path, create, readonly)
else:
self._open_unix(path, create)
self._open_unix(path, create, readonly)
def open_and_read(self, path: Path, create: bool = False) -> bytes:
"""Open *path* with exclusive lock and read all content.
def open_and_read(
self, path: Path, create: bool = False, readonly: bool = False
) -> bytes:
"""Open *path* and read all content.
Combined operation for efficient use with asyncio.to_thread().
"""
self.open(path, create=create)
self.open(path, create=create, readonly=readonly)
return self.read()
def read(self) -> bytes:
@@ -188,20 +193,24 @@ class LockedFile:
# -- Unix ----------------------------------------------------------------
def _open_unix(self, path: Path, create: bool) -> None:
flags = os.O_RDWR | (os.O_CREAT if create else 0)
def _open_unix(self, path: Path, create: bool, readonly: bool) -> None:
if readonly:
flags = os.O_RDONLY
else:
flags = os.O_RDWR | (os.O_CREAT if create else 0)
try:
fd = os.open(path, flags, 0o666)
except FileNotFoundError:
_fatal(f"Database file not found: {path.resolve()}", db_path=path)
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
os.close(fd)
_fatal(
f"{path.resolve()}: database already locked by another instance",
db_path=path,
)
if not readonly:
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
os.close(fd)
_fatal(
f"{path.resolve()}: database already locked by another instance",
db_path=path,
)
self._fd = fd
def _read_unix(self) -> bytes:
@@ -220,12 +229,19 @@ class LockedFile:
# -- Windows -------------------------------------------------------------
def _open_win32(self, path: Path, create: bool) -> None:
disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING
def _open_win32(self, path: Path, create: bool, readonly: bool) -> None:
if readonly:
disposition = _OPEN_EXISTING
access = _GENERIC_READ
share = _FILE_SHARE_READ | _FILE_SHARE_WRITE
else:
disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING
access = _GENERIC_READ | _GENERIC_WRITE
share = _FILE_SHARE_READ
handle = _kernel32.CreateFileW(
str(path),
_GENERIC_READ | _GENERIC_WRITE,
_FILE_SHARE_READ,
access,
share,
None,
disposition,
_FILE_ATTRIBUTE_NORMAL,
+52 -7
View File
@@ -1,10 +1,11 @@
"""Kanta DB main public API"""
from __future__ import annotations
import logging
from datetime import datetime
from pathlib import Path
from types import ModuleType
from typing import Any, Generic, TypeVar
from types import ModuleType, SimpleNamespace
from typing import Generic, TypeVar
from kanta.kantaimpl import KantaImpl
from kanta.serialization import JsonSerializer, Serializer
@@ -50,7 +51,6 @@ class Kanta(Generic[T]):
*,
type: type[T] | None = None,
migrations: ModuleType | str | None = None,
migration_ctx: Any | None = None,
serializer: Serializer | None = None,
flush_interval: float = 0.1,
):
@@ -61,7 +61,6 @@ class Kanta(Generic[T]):
data: Caller-owned root msgspec.Struct state instance.
type: Optional explicit root type. Defaults to ``type(data)``.
migrations: Optional migrations module object or import path.
migration_ctx: Optional context object passed to migration functions.
flush_interval: Background flush interval in seconds.
serializer: Optional serializer implementation.
@@ -78,7 +77,6 @@ class Kanta(Generic[T]):
data=data,
type=data_type,
migrations=migrations,
migration_ctx=migration_ctx,
flush_interval=flush_interval,
kanta=self,
)
@@ -127,6 +125,15 @@ class Kanta(Generic[T]):
"""
return self._impl.filename
@property
def ctx(self) -> SimpleNamespace:
"""User-writable context namespace.
Migration functions receive the ``Kanta`` instance and can read or
mutate ``kanta.ctx`` during migrations.
"""
return self._impl.ctx
@property
def mtime(self) -> datetime | None:
"""Last modification time carried forward from change records.
@@ -138,7 +145,13 @@ class Kanta(Generic[T]):
"""
return self._impl.mtime
async def open(self, *, create: bool = True) -> 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
@@ -147,6 +160,14 @@ class Kanta(Generic[T]):
Args:
create: Whether to create the database file when missing.
If False, opening fails when the file does not exist or is empty.
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 migration logging. ``True`` (default) uses the
``kanta.migrations`` logger. ``False`` suppresses the default
migration log. A :class:`~logging.Logger` instance writes
default migration 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.
@@ -154,7 +175,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)
await self._impl.open(create=create, readonly=readonly, log=log)
async def __aenter__(self) -> Kanta[T]:
"""Enter async context manager and open the database.
@@ -230,6 +251,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.
@@ -257,6 +296,7 @@ class Kanta(Generic[T]):
*,
user: str | None = None,
mtime: bool | datetime = True,
log: bool | logging.Logger = True,
):
"""Create a transactional mutation context manager.
@@ -271,6 +311,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.changes`` 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.
@@ -285,4 +329,5 @@ class Kanta(Generic[T]):
action,
user=user,
mtime=mtime,
log=log,
)
+149 -33
View File
@@ -7,11 +7,13 @@ import copy
import importlib
import logging
from datetime import UTC, datetime
from types import SimpleNamespace
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 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
@@ -27,22 +29,23 @@ 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.migration_ctx = kwargs.pop("migration_ctx", 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
self.opened = False
self.readonly = False
self.bootstrap_action = "bootstrap"
self.bootstrap_user: str | None = None
self.bootstrap_mtime: bool | datetime = True
@@ -53,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,
@@ -75,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) -> 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(
@@ -84,12 +142,17 @@ class KantaImpl(PersistenceMixin, Generic[T]):
action="open",
)
self.readonly = readonly
existed_before_open = self.filename.exists()
# Read-only mode never creates the file.
open_create = create and not readonly
content = await asyncio.to_thread(
self.file.open_and_read,
self.filename,
create=create,
create=open_create,
readonly=readonly,
)
if not create and (not existed_before_open or not content):
@@ -105,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(
@@ -134,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(
rr.state, rr.version, self.migration_ctx
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,
@@ -149,27 +236,56 @@ class KantaImpl(PersistenceMixin, Generic[T]):
self.version = rr.version
self.mtime = rr.m
normalized = struct_to_dict(self.data, serializer=self.serializer)
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
)
elif self.callback_registry.has("bootstrap"):
try:
await self.callback_registry.invoke(
"bootstrap",
InjectionContext(data=self.data, kanta=self._kanta),
if self.readonly:
self.statedict = copy.deepcopy(normalized)
else:
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",
db_path=self.filename,
action="open",
)
else:
try:
if self.callback_registry.has("bootstrap"):
await self.callback_registry.invoke(
"bootstrap",
InjectionContext(data=self.data, kanta=self._kanta),
)
self.statedict = {}
current = struct_to_dict(self.data, serializer=self.serializer)
self.queue_change(
self.bootstrap_action,
current,
user=self.bootstrap_user,
mtime=self.bootstrap_mtime,
force=True,
)
except Exception:
self.opened = False
self.file.close()
try:
await asyncio.to_thread(self.filename.unlink, missing_ok=True)
@@ -177,9 +293,8 @@ class KantaImpl(PersistenceMixin, Generic[T]):
pass
raise
self.opened = True
self.background_task = asyncio.create_task(self._background_loop())
if not self.readonly:
self.background_task = asyncio.create_task(self._background_loop())
async def close(self) -> None:
"""Stop the background task, flush pending changes, and release the file lock."""
@@ -196,7 +311,8 @@ class KantaImpl(PersistenceMixin, Generic[T]):
# Always run a final flush in case the background task never reached
# its cancellation handler.
await self.flush()
if not self.readonly:
await self.flush()
self.file.close()
self.opened = False
+15 -9
View File
@@ -10,7 +10,8 @@ import sys
from collections.abc import Callable
from typing import Any
logger = logging.getLogger("kanta.changes")
changes_logger = logging.getLogger("kanta.changes")
migration_logger = logging.getLogger("kanta.migrations")
# Pattern to match control characters and bidirectional overrides
_UNSAFE_CHARS = re.compile(
@@ -274,6 +275,9 @@ def log_change(
user: str | None = None,
previous: dict | None = None,
logfmt: Callable[[Any, str], str | None] | None = None,
*,
logger: logging.Logger = changes_logger,
level: int = logging.INFO,
) -> None:
"""Log a database change with pretty-printed diff.
@@ -283,27 +287,29 @@ 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.changes`` 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:
if not changes_logger.handlers:
handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(handler)
logger.setLevel(logging.INFO)
logger.propagate = False
changes_logger.addHandler(handler)
changes_logger.setLevel(logging.INFO)
changes_logger.propagate = False
-117
View File
@@ -1,117 +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 logging
from types import ModuleType
from typing import Any
import msgspec
_logger = logging.getLogger(__name__)
class MigrationCtx(msgspec.Struct, omit_defaults=True):
"""Context passed to each migration function.
Subclass or replace this with your own context type.
"""
pass
class MigrationRegistry:
"""Registry of schema migration functions.
Usage::
registry = MigrationRegistry()
@registry.register
def migrate_v1(d: dict, ctx: MigrationCtx) -> None:
d.setdefault("version", 1)
new_version = registry.apply(state, current_version=0)
Or load from a module::
registry = MigrationRegistry.from_module("myapp.migrations")
new_version = registry.apply(state, current_version=0)
"""
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)
def apply(
self,
data_dict: dict[str, Any],
current_version: int,
ctx: MigrationCtx | None = None,
*,
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})"
)
fn(data_dict, ctx or MigrationCtx())
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)
+24 -2
View File
@@ -39,6 +39,7 @@ class PersistenceMixin:
flush_interval: float
version: int
opened: bool
readonly: bool
mtime: datetime | None
def __init__(self, **kwargs: Any) -> None:
@@ -68,6 +69,8 @@ class PersistenceMixin:
async def _background_loop(self) -> None:
"""Background task that periodically flushes changes to disk."""
if self.readonly:
return
while True:
try:
await asyncio.sleep(self.flush_interval)
@@ -106,6 +109,7 @@ class PersistenceMixin:
*,
user: str | None = None,
mtime: bool | datetime = True,
force: bool = False,
) -> ChangeRecord | None:
"""Queue a change record internally (thread-safe).
@@ -118,9 +122,11 @@ class PersistenceMixin:
previous modification time remains in effect; this is used for
system operations that are not considered modifications. A
:class:`~datetime.datetime` value sets ``m`` to that explicit time.
force: If ``True``, queue the record even when the diff is empty.
Returns:
The queued :class:`ChangeRecord`, or ``None`` if the diff was empty.
The queued :class:`ChangeRecord`, or ``None`` if the diff was empty
and *force* is ``False``.
"""
now = datetime.now(UTC)
@@ -135,7 +141,9 @@ class PersistenceMixin:
diff = compute_diff(self.statedict, current)
if not diff:
return None
if not force:
return None
diff = {}
record = ChangeRecord(
ts=now,
@@ -160,6 +168,13 @@ class PersistenceMixin:
action="flush_sync",
)
if self.readonly:
raise DataIntegrityError(
"Cannot flush in read-only mode",
db_path=self.filename,
action="flush_sync",
)
if self.flush_failed:
return
@@ -207,6 +222,13 @@ class PersistenceMixin:
action="flush",
)
if self.readonly:
raise DataIntegrityError(
"Cannot flush in read-only mode",
db_path=self.filename,
action="flush",
)
if self.flush_failed:
return
+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):
+19 -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, changes_logger, log_change
from kanta.serialization import restore_data_in_place, struct_to_dict
_logger = logging.getLogger(__name__)
@@ -22,8 +22,16 @@ 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:
raise DataIntegrityError(
"Cannot start transaction in read-only mode",
db_path=impl.filename,
action=action,
)
if impl.in_transaction:
raise RuntimeError(
"Nested or simultaneous transactions are not supported "
@@ -73,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 changes_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:
+24 -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):
@@ -70,6 +70,18 @@ def change_actions(path: Path, format_config) -> list[str]:
return actions
def read_changes(path: Path, format_config) -> list[ChangeRecord]:
_, serializer_cls = format_config
serializer = serializer_cls()
framer = serializer.framer_cls()
records: list[ChangeRecord] = []
for is_snapshot, payload, _, _ in framer.iter_records(path.read_bytes(), 0):
if is_snapshot:
continue
records.append(serializer.decode(payload, type=ChangeRecord))
return records
def make_migrations_module(name: str, fn_name: str, fn):
mod = ModuleType(name)
mod.__dict__[fn_name] = fn
@@ -77,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
+49 -5
View File
@@ -1,4 +1,4 @@
from typing import Any
from typing import Any, Optional, Union
import pytest
@@ -49,16 +49,60 @@ def test_logfmt_requires_value_annotation(tmp_path, format_config):
return None
def test_logfmt_requires_return_annotation(tmp_path, format_config):
def test_logfmt_allows_missing_return_annotation(tmp_path, format_config):
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
with pytest.raises(TypeError, match="must annotate its return"):
@kanta.logfmt
def resolve_names(value: str, current: DictPost):
return None
@kanta.logfmt
def resolve_names(value: str, current: DictPost):
def test_logfmt_class_allows_missing_return_annotation(tmp_path, format_config):
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
@kanta.logfmt
class UserLogFmt(LogFmt):
def resolve(self, value: str, path: str):
return None
# fmt: off
def test_logfmt_accepts_optional_return_typing_forms(tmp_path, format_config):
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
@kanta.logfmt
def resolve_optional(value: str) -> Optional[str]: # noqa: UP007
return value
@kanta.logfmt
def resolve_union(value: str) -> Union[str, None]: # noqa: UP007
return value
@kanta.logfmt
def resolve_pipe(value: "str") -> "str | None":
return value
def test_logfmt_class_accepts_optional_return_typing_forms(tmp_path, format_config):
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
@kanta.logfmt
class OptionalStyle(LogFmt):
def resolve(self, value: str, path: str) -> Optional[str]: # noqa: UP007
return value
@kanta.logfmt
class UnionStyle(LogFmt):
def resolve(self, value: str, path: str) -> Union[str, None]: # noqa: UP007
return value
@kanta.logfmt
class StringStyle(LogFmt):
def resolve(self, value: "str", path: "str") -> "str | None":
return value
# fmt: on
def test_logfmt_rejects_async_callback(tmp_path, format_config):
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+261 -2
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,6 +19,9 @@ from .support import (
change_actions,
fixed_change,
make_kanta,
make_migrations_module,
read_changes,
read_last_snapshot,
seed_single_change,
)
@@ -30,6 +35,63 @@ async def test_load_empty(tmp_path, format_config):
await kanta.close()
@pytest.mark.asyncio
async def test_new_file_writes_bootstrap_record_without_handlers(
tmp_path, format_config
):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
await kanta.open()
await kanta.close()
records = read_changes(path, format_config)
assert len(records) == 1
assert records[0].a == "bootstrap"
assert records[0].diff == {"$replace": {"users": {}, "counter": 0}}
@pytest.mark.asyncio
async def test_new_file_persists_initial_state_for_roundtrip(tmp_path, format_config):
path = tmp_path / "test.db"
kanta = make_kanta(
path, Data(counter=5, users={"alice": User(name="Alice")}), format_config
)
await kanta.open()
await kanta.close()
records = read_changes(path, format_config)
assert len(records) == 1
assert records[0].a == "bootstrap"
assert records[0].diff == {
"$replace": {"users": {"alice": {"name": "Alice", "age": 0}}, "counter": 5}
}
kanta2 = make_kanta(path, Data, format_config)
await kanta2.open()
assert kanta2.data.counter == 5
assert kanta2.data.users["alice"].name == "Alice"
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"
@@ -404,7 +466,7 @@ async def test_migrations_from_module(tmp_path, format_config):
mod = type(sys)("test_migrations")
def migrate_v1(d, ctx):
def migrate_v1(d, kanta):
d["version"] = 1
mod.__dict__["migrate_v1"] = migrate_v1
@@ -434,6 +496,203 @@ 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.migrations"):
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.migrations"):
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_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.migrations"):
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.changes"):
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"
@@ -514,7 +773,7 @@ async def test_migrations_from_module_path(tmp_path, format_config):
module_name = "test_migrations_path"
mod = type(sys)(module_name)
def migrate_v1(d, ctx):
def migrate_v1(d, kanta):
d["counter"] = 2
mod.__dict__["migrate_v1"] = migrate_v1
+3 -4
View File
@@ -1,16 +1,15 @@
import logging
from kanta.logging import configure_logging, log_change
from kanta.logging import logger
from kanta.logging import changes_logger, configure_logging, log_change
def test_configure_logging():
configure_logging()
assert logger.level == logging.INFO
assert changes_logger.level == logging.INFO
def test_log_change_no_diff(capsys):
logger.handlers.clear()
changes_logger.handlers.clear()
configure_logging()
log_change("test", {})
captured = capsys.readouterr()
+171 -16
View File
@@ -1,53 +1,208 @@
from types import ModuleType
from types import ModuleType, SimpleNamespace
from kanta.migrate import MigrationRegistry
import pytest
from kanta.exceptions import DatabaseError
from kanta.migrations import Migrations
class _DummyKanta:
def __init__(self):
self.ctx = SimpleNamespace()
def test_register_and_apply():
reg = MigrationRegistry()
reg = Migrations()
kanta = _DummyKanta()
@reg.register
def migrate_v1(d, ctx):
def migrate_v1(d, kanta):
d["version"] = 1
@reg.register
def migrate_v2(d, ctx):
def migrate_v2(d, kanta):
d["version"] = 2
state = {}
new_ver = reg.apply(state, current_version=0, 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
def migrate_v1(d, ctx):
def migrate_v1(d, kanta):
d["x"] = 1
state = {"x": 1}
new_ver = reg.apply(state, current_version=1, silent=True)
assert new_ver == 1
result = reg.apply(state, current_version=1, kanta=kanta)
assert result.version == 1
def test_from_module():
mod = ModuleType("fake_migrations")
kanta = _DummyKanta()
def migrate_v1(d, ctx):
def migrate_v1(d, kanta):
d["v"] = 1
def migrate_v2(d, ctx):
def migrate_v2(d, kanta):
d["v"] = 2
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, 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 = Migrations()
kanta = _DummyKanta()
@reg.register
def migrate_v1(d, kanta):
kanta.ctx.source = "migration"
d["source"] = kanta.ctx.source
state = {}
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 = Migrations()
kanta = _DummyKanta()
@reg.register
def migrate_v1(d):
d["x"] = 1
state = {}
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"
+3 -2
View File
@@ -82,8 +82,9 @@ async def test_transaction_mtime_false_preserves_mtime(tmp_path, format_config):
continue
records.append(serializer.decode(payload, type=ChangeRecord))
assert records[0].m == first_m
assert records[1].m is None
assert records[0].a == "bootstrap"
assert records[1].m == first_m
assert records[2].m is None
assert kanta.mtime == first_m
+153
View File
@@ -0,0 +1,153 @@
"""Tests for Kanta read-only mode."""
import pytest
from kanta.exceptions import DataIntegrityError, FileLockError
from kanta.serialization import struct_to_dict
from .support import (
Data,
EvolvableDataV2,
fixed_change,
make_kanta,
make_migrations_module,
seed_single_change,
)
@pytest.mark.asyncio
async def test_readonly_opens_existing_database(tmp_path, format_config):
path = tmp_path / "test.db"
seed_single_change(path, fixed_change("seed", {"counter": 7}), format_config)
kanta = make_kanta(path, Data, format_config)
await kanta.open(readonly=True)
assert isinstance(kanta.data, Data)
assert kanta.data.counter == 7
assert kanta._impl.readonly is True
assert kanta._impl.background_task is None
await kanta.close()
@pytest.mark.asyncio
async def test_readonly_missing_file_fails(tmp_path, format_config):
path = tmp_path / "missing.db"
kanta = make_kanta(path, Data, format_config)
with pytest.raises(FileLockError):
await kanta.open(readonly=True)
assert not path.exists()
@pytest.mark.asyncio
async def test_readonly_empty_file_fails(tmp_path, format_config):
path = tmp_path / "empty.db"
path.touch()
kanta = make_kanta(path, Data, format_config)
with pytest.raises(DataIntegrityError, match="empty"):
await kanta.open(readonly=True)
@pytest.mark.asyncio
async def test_readonly_transaction_fails(tmp_path, format_config):
path = tmp_path / "test.db"
seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
kanta = make_kanta(path, Data, format_config)
await kanta.open(readonly=True)
with pytest.raises(DataIntegrityError, match="read-only"):
with kanta.transaction(action="inc") as data:
data.counter = 2
# In-memory state must remain unchanged.
assert kanta.data.counter == 1
await kanta.close()
@pytest.mark.asyncio
async def test_readonly_flush_fails(tmp_path, format_config):
path = tmp_path / "test.db"
seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
kanta = make_kanta(path, Data, format_config)
await kanta.open(readonly=True)
with pytest.raises(DataIntegrityError, match="read-only"):
await kanta.flush()
await kanta.close()
@pytest.mark.asyncio
async def test_readonly_create_true_does_not_create_file(tmp_path, format_config):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
with pytest.raises(FileLockError):
await kanta.open(create=True, readonly=True)
assert not path.exists()
@pytest.mark.asyncio
async def test_readonly_does_not_persist_changes(tmp_path, format_config):
path = tmp_path / "test.db"
seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
original_content = path.read_bytes()
kanta = make_kanta(path, Data, format_config)
await kanta.open(readonly=True)
await kanta.close()
assert path.read_bytes() == original_content
@pytest.mark.asyncio
async def test_readonly_runs_migrations(tmp_path, format_config):
path = tmp_path / "test.db"
seed_single_change(
path,
fixed_change("seed", {"counter": 1}, version=0),
format_config,
)
def migrate_v1(data, kanta):
data.setdefault("enabled", True)
migrations = make_migrations_module("readonly_migrations", "migrate_v1", migrate_v1)
kanta = make_kanta(path, EvolvableDataV2, format_config, migrations=migrations)
await kanta.open(readonly=True)
assert kanta.data.counter == 1
# Migration ran in memory even though no change was persisted.
assert struct_to_dict(kanta.data, serializer=kanta._impl.serializer) == {
"counter": 1,
"enabled": True,
}
assert not kanta._impl.pending_changes
await kanta.close()
@pytest.mark.asyncio
async def test_readwrite_and_readonly_can_open_together(tmp_path, format_config):
path = tmp_path / "test.db"
seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
rw = make_kanta(path, Data, format_config)
await rw.open()
ro = make_kanta(path, Data, format_config)
await ro.open(readonly=True)
assert rw.data.counter == 1
assert ro.data.counter == 1
await ro.close()
await rw.close()
+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