Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3a56bfbb10 | ||
|
|
55fa475a13 | ||
|
|
753b7eba86 | ||
|
|
42789e6619 |
+10
-1
@@ -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
@@ -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,
|
||||
)
|
||||
|
||||
+135
-19
@@ -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.migrations import Migrations
|
||||
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
|
||||
@@ -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, 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(
|
||||
@@ -110,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(
|
||||
@@ -139,13 +200,33 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
||||
cause_type=type(e).__name__,
|
||||
) from e
|
||||
|
||||
migrations_ran = False
|
||||
migration_result = None
|
||||
state_before_migrations = None
|
||||
previous_version = rr.version
|
||||
if self.migrations is not None:
|
||||
previous_version = rr.version
|
||||
rr.version = self.migrations.apply(rr.state, rr.version, self._kanta)
|
||||
migrations_ran = rr.version != previous_version
|
||||
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,
|
||||
@@ -154,24 +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:
|
||||
if migrations_ran:
|
||||
if migrations_ran and migration_state_changed:
|
||||
self.queue_change(
|
||||
f"migrate:v{self.version}",
|
||||
self.statedict,
|
||||
rr.state,
|
||||
mtime=False,
|
||||
force=True,
|
||||
)
|
||||
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
|
||||
)
|
||||
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",
|
||||
@@ -188,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)
|
||||
@@ -203,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
@@ -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)
|
||||
|
||||
+44
-14
@@ -9,19 +9,38 @@ from __future__ import annotations
|
||||
import copy
|
||||
import importlib
|
||||
import inspect
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from types import ModuleType
|
||||
from typing import Any
|
||||
|
||||
from kanta.diff import compute_diff
|
||||
from kanta.exceptions import DatabaseError
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
# Cache registries by imported module object so that many Kanta instances using
|
||||
# the same migrations module do not re-scan it each time.
|
||||
_module_registry_cache: dict[ModuleType, Migrations] = {}
|
||||
|
||||
|
||||
@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.
|
||||
|
||||
@@ -38,12 +57,13 @@ class Migrations:
|
||||
def migrate_v2(d: dict) -> None:
|
||||
d.setdefault("version", 2)
|
||||
|
||||
new_version = migrations.apply(state, current_version=0, kanta=kanta)
|
||||
result = migrations.apply(state, current_version=0, kanta=kanta)
|
||||
new_version = result.version
|
||||
|
||||
Or load from a module::
|
||||
|
||||
migrations = Migrations.from_module("myapp.migrations")
|
||||
new_version = migrations.apply(state, current_version=0, kanta=kanta)
|
||||
result = migrations.apply(state, current_version=0, kanta=kanta)
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
@@ -117,9 +137,7 @@ class Migrations:
|
||||
data_dict: dict[str, Any],
|
||||
current_version: int,
|
||||
kanta: Any,
|
||||
*,
|
||||
silent: bool = False,
|
||||
) -> int:
|
||||
) -> MigrationResult:
|
||||
"""Apply pending migrations to *data_dict* in place.
|
||||
|
||||
Missing intermediate migration steps are silently skipped.
|
||||
@@ -128,7 +146,8 @@ class Migrations:
|
||||
DatabaseError: If the database version is newer than the highest
|
||||
supported version or older than the minimum supported version.
|
||||
|
||||
Returns the new version after all migrations.
|
||||
Returns a :class:`MigrationResult` describing the new version and every
|
||||
migration that ran.
|
||||
"""
|
||||
if current_version > self.dbver:
|
||||
raise DatabaseError(
|
||||
@@ -141,14 +160,25 @@ class Migrations:
|
||||
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) if not silent else None
|
||||
before = copy.deepcopy(data_dict)
|
||||
self._call_migration(fn, data_dict, kanta)
|
||||
current_version = version
|
||||
if not silent and before != data_dict:
|
||||
desc = (fn.__doc__ or fn.__name__).split("\n")[0].rstrip(".")
|
||||
_logger.info("Applied migration %s: %s", fn.__name__, desc)
|
||||
return current_version
|
||||
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)
|
||||
|
||||
+8
-7
@@ -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
@@ -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
@@ -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
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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 (
|
||||
@@ -19,6 +21,7 @@ from .support import (
|
||||
make_kanta,
|
||||
make_migrations_module,
|
||||
read_changes,
|
||||
read_last_snapshot,
|
||||
seed_single_change,
|
||||
)
|
||||
|
||||
@@ -70,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"
|
||||
@@ -475,7 +497,9 @@ async def test_msgspec_normalization_logs_migration(tmp_path, format_config):
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_empty_migration_is_recorded_and_not_reapplied(tmp_path, format_config):
|
||||
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
|
||||
@@ -491,26 +515,249 @@ async def test_empty_migration_is_recorded_and_not_reapplied(tmp_path, format_co
|
||||
kanta = make_kanta(path, Data, format_config, migrations=mod)
|
||||
await kanta.open()
|
||||
assert kanta.version == 1
|
||||
await kanta.flush()
|
||||
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 len(migration_records) == 1
|
||||
assert migration_records[0].v == 1
|
||||
assert migration_records[0].diff == {}
|
||||
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 len([r for r in records2 if r.a.startswith("migrate")]) == 1
|
||||
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
@@ -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()
|
||||
|
||||
+38
-34
@@ -1,4 +1,3 @@
|
||||
import logging
|
||||
from types import ModuleType, SimpleNamespace
|
||||
|
||||
import pytest
|
||||
@@ -25,8 +24,8 @@ 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
|
||||
|
||||
|
||||
@@ -39,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():
|
||||
@@ -60,8 +59,8 @@ def test_from_module():
|
||||
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
|
||||
|
||||
|
||||
@@ -75,8 +74,8 @@ 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"
|
||||
|
||||
@@ -90,8 +89,8 @@ 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
|
||||
|
||||
|
||||
@@ -107,7 +106,7 @@ def test_version_too_new():
|
||||
DatabaseError,
|
||||
match="Database version v2 is newer than the highest supported version v1",
|
||||
):
|
||||
reg.apply({}, current_version=2, kanta=kanta, silent=True)
|
||||
reg.apply({}, current_version=2, kanta=kanta)
|
||||
|
||||
|
||||
def test_version_too_old():
|
||||
@@ -122,7 +121,7 @@ def test_version_too_old():
|
||||
DatabaseError,
|
||||
match="Database version v1 is older than the minimum supported version v2",
|
||||
):
|
||||
reg.apply({}, current_version=1, kanta=kanta, silent=True)
|
||||
reg.apply({}, current_version=1, kanta=kanta)
|
||||
|
||||
|
||||
def test_missing_middle_migration_is_skipped():
|
||||
@@ -138,8 +137,8 @@ def test_missing_middle_migration_is_skipped():
|
||||
d["y"] = 3
|
||||
|
||||
state = {"x": 1}
|
||||
new_ver = reg.apply(state, current_version=1, kanta=kanta, silent=True)
|
||||
assert new_ver == 3
|
||||
result = reg.apply(state, current_version=1, kanta=kanta)
|
||||
assert result.version == 3
|
||||
assert state["x"] == 1
|
||||
assert state["y"] == 3
|
||||
|
||||
@@ -153,12 +152,12 @@ def test_old_migrations_deleted_current_supported():
|
||||
d["x"] = 3
|
||||
|
||||
state = {"x": 2}
|
||||
new_ver = reg.apply(state, current_version=2, kanta=kanta, silent=True)
|
||||
assert new_ver == 3
|
||||
result = reg.apply(state, current_version=2, kanta=kanta)
|
||||
assert result.version == 3
|
||||
assert state["x"] == 3
|
||||
|
||||
|
||||
def test_migration_log_only_when_changed(caplog):
|
||||
def test_apply_returns_change_information():
|
||||
reg = Migrations()
|
||||
kanta = _DummyKanta()
|
||||
|
||||
@@ -177,28 +176,33 @@ def test_migration_log_only_when_changed(caplog):
|
||||
"""Set y."""
|
||||
d["y"] = 3
|
||||
|
||||
with caplog.at_level(logging.INFO, logger="kanta.migrations"):
|
||||
reg.apply({}, current_version=0, kanta=kanta)
|
||||
result = reg.apply({}, current_version=0, kanta=kanta)
|
||||
assert result.version == 3
|
||||
assert len(result.migrations) == 3
|
||||
|
||||
messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
|
||||
assert len(messages) == 2
|
||||
assert "migrate_v1" in messages[0]
|
||||
assert "Set x" in messages[0]
|
||||
assert "migrate_v3" in messages[1]
|
||||
assert "Set y" in messages[1]
|
||||
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_no_op_migration_produces_no_log(caplog):
|
||||
def test_description_defaults_to_version_when_no_docstring():
|
||||
reg = Migrations()
|
||||
kanta = _DummyKanta()
|
||||
|
||||
@reg.register
|
||||
def migrate_v1(d):
|
||||
"""No-op."""
|
||||
pass
|
||||
d["x"] = 1
|
||||
|
||||
with caplog.at_level(logging.INFO, logger="kanta.migrations"):
|
||||
reg.apply({}, current_version=0, kanta=kanta)
|
||||
|
||||
info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
|
||||
assert not info_messages
|
||||
result = reg.apply({}, current_version=0, kanta=kanta)
|
||||
assert result.migrations[0].description == "v1"
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user