Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3a56bfbb10 | ||
|
|
55fa475a13 | ||
|
|
753b7eba86 | ||
|
|
42789e6619 | ||
|
|
c4726e6728 | ||
|
|
66e92739ab | ||
|
|
4dc2f0648e | ||
|
|
bec4635460 | ||
|
|
c04a245366 |
@@ -53,10 +53,13 @@ asyncio.run(main())
|
|||||||
|
|
||||||
## Bootstrap and Open Modes
|
## Bootstrap and Open Modes
|
||||||
|
|
||||||
Kanta supports open-time bootstrap callbacks for initializing a brand-new
|
When `open()` creates a brand-new database, it always writes a single bootstrap
|
||||||
database before `open()` returns.
|
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
|
```python
|
||||||
kanta = Kanta("data.kantadb", Data())
|
kanta = Kanta("data.kantadb", Data())
|
||||||
@@ -76,9 +79,10 @@ async def bootstrap_async(data) -> None:
|
|||||||
data.counter = 1
|
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,
|
- they run in registration order,
|
||||||
- exactly one bootstrap change record is queued,
|
|
||||||
- bootstrap metadata (`action`, `user`, `mtime`) is taken from the last
|
- bootstrap metadata (`action`, `user`, `mtime`) is taken from the last
|
||||||
registration.
|
registration.
|
||||||
|
|
||||||
@@ -94,6 +98,18 @@ await kanta.open(create=False)
|
|||||||
With `create=False`, open fails if the database file does not exist or is
|
With `create=False`, open fails if the database file does not exist or is
|
||||||
empty.
|
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 Error Handlers
|
||||||
|
|
||||||
Fatal background write errors can be observed with a decorator:
|
Fatal background write errors can be observed with a decorator:
|
||||||
|
|||||||
+13
-1
@@ -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()` (default) creates the database file if missing.
|
||||||
- `await kanta.open(create=False)` fails when the file is missing or empty.
|
- `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
|
### Callbacks
|
||||||
|
|
||||||
@@ -127,7 +133,11 @@ when they have a default value.
|
|||||||
|
|
||||||
#### Bootstrap Callbacks
|
#### 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:
|
- Register callbacks via:
|
||||||
- `@kanta.bootstrap`
|
- `@kanta.bootstrap`
|
||||||
- `@kanta.bootstrap(action=..., user=..., mtime=...)`
|
- `@kanta.bootstrap(action=..., user=..., mtime=...)`
|
||||||
@@ -140,6 +150,8 @@ when they have a default value.
|
|||||||
- exactly one bootstrap `ChangeRecord` is queued,
|
- exactly one bootstrap `ChangeRecord` is queued,
|
||||||
- bootstrap metadata (`action`, `user`, `mtime`) is taken from the last
|
- bootstrap metadata (`action`, `user`, `mtime`) is taken from the last
|
||||||
callback registration.
|
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,
|
- If any bootstrap callback raises, Kanta closes and removes the database file,
|
||||||
then re-raises the exception.
|
then re-raises the exception.
|
||||||
|
|
||||||
|
|||||||
+15
-13
@@ -12,11 +12,13 @@ and receive the value plus an optional ``path`` string. They return
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import inspect
|
import inspect
|
||||||
|
import types
|
||||||
from collections.abc import Callable
|
from collections.abc import Callable
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from typing import Annotated, Any, Union, get_args, get_origin
|
from typing import Annotated, Any, Union, get_args, get_origin
|
||||||
|
|
||||||
from kanta.exceptions import DatabaseError
|
from kanta.exceptions import DatabaseError
|
||||||
|
from kanta.migrations import MigrationResult
|
||||||
|
|
||||||
DictPre = Annotated[dict, "pre"]
|
DictPre = Annotated[dict, "pre"]
|
||||||
DictPost = Annotated[dict, "post"]
|
DictPost = Annotated[dict, "post"]
|
||||||
@@ -58,6 +60,7 @@ class InjectionContext:
|
|||||||
error: DatabaseError | None = None
|
error: DatabaseError | None = None
|
||||||
previous_state: dict | None = None
|
previous_state: dict | None = None
|
||||||
current_state: dict | None = None
|
current_state: dict | None = None
|
||||||
|
migration_result: MigrationResult | None = None
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
@@ -97,6 +100,7 @@ class CallbackRegistry:
|
|||||||
self._callbacks: dict[str, list[_CallbackRegistration]] = {
|
self._callbacks: dict[str, list[_CallbackRegistration]] = {
|
||||||
"bootstrap": [],
|
"bootstrap": [],
|
||||||
"fatal_error": [],
|
"fatal_error": [],
|
||||||
|
"logmigr": [],
|
||||||
}
|
}
|
||||||
self._logfmt_callbacks: list[_LogFmtFunctionSpec | _LogFmtClassSpec] = []
|
self._logfmt_callbacks: list[_LogFmtFunctionSpec | _LogFmtClassSpec] = []
|
||||||
|
|
||||||
@@ -317,11 +321,7 @@ class CallbackRegistry:
|
|||||||
f"Allowed: str path, {self._allowed_message('logfmt')}"
|
f"Allowed: str path, {self._allowed_message('logfmt')}"
|
||||||
)
|
)
|
||||||
|
|
||||||
if sig.return_annotation is inspect.Signature.empty:
|
if sig.return_annotation is not 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)
|
return_ann = self._resolve_raw_annotation(sig.return_annotation, callback)
|
||||||
if not self._is_optional_str(return_ann):
|
if not self._is_optional_str(return_ann):
|
||||||
raise TypeError(
|
raise TypeError(
|
||||||
@@ -416,11 +416,7 @@ class CallbackRegistry:
|
|||||||
f"logfmt class {cls.__name__}.resolve must accept a 'path: str' parameter"
|
f"logfmt class {cls.__name__}.resolve must accept a 'path: str' parameter"
|
||||||
)
|
)
|
||||||
|
|
||||||
if resolve_sig.return_annotation is inspect.Signature.empty:
|
if resolve_sig.return_annotation is not 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(
|
return_ann = self._resolve_raw_annotation(
|
||||||
resolve_sig.return_annotation, resolve
|
resolve_sig.return_annotation, resolve
|
||||||
)
|
)
|
||||||
@@ -453,10 +449,12 @@ class CallbackRegistry:
|
|||||||
return kind == "logfmt"
|
return kind == "logfmt"
|
||||||
if bare is DatabaseError:
|
if bare is DatabaseError:
|
||||||
return kind == "fatal_error"
|
return kind == "fatal_error"
|
||||||
|
if bare is MigrationResult:
|
||||||
|
return kind == "logmigr"
|
||||||
if self._data_type is not None and bare is self._data_type:
|
if self._data_type is not None and bare is self._data_type:
|
||||||
return kind == "bootstrap"
|
return kind == "bootstrap"
|
||||||
if self._kanta_class is not None and bare is self._kanta_class:
|
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
|
return False
|
||||||
|
|
||||||
def _allowed_message(self, kind: str) -> str:
|
def _allowed_message(self, kind: str) -> str:
|
||||||
@@ -469,6 +467,8 @@ class CallbackRegistry:
|
|||||||
parts.append(self._kanta_class.__name__)
|
parts.append(self._kanta_class.__name__)
|
||||||
if kind == "fatal_error":
|
if kind == "fatal_error":
|
||||||
parts.append("DatabaseError")
|
parts.append("DatabaseError")
|
||||||
|
if kind == "logmigr":
|
||||||
|
parts.append("MigrationResult")
|
||||||
if kind == "logfmt":
|
if kind == "logfmt":
|
||||||
parts.append("Annotated[dict, 'pre']")
|
parts.append("Annotated[dict, 'pre']")
|
||||||
parts.append("Annotated[dict, 'post']")
|
parts.append("Annotated[dict, 'post']")
|
||||||
@@ -482,6 +482,8 @@ class CallbackRegistry:
|
|||||||
return ctx.current_state
|
return ctx.current_state
|
||||||
if bare is DatabaseError:
|
if bare is DatabaseError:
|
||||||
return ctx.error
|
return ctx.error
|
||||||
|
if bare is MigrationResult:
|
||||||
|
return ctx.migration_result
|
||||||
if self._data_type is not None and bare is self._data_type:
|
if self._data_type is not None and bare is self._data_type:
|
||||||
return ctx.data
|
return ctx.data
|
||||||
if self._kanta_class is not None and bare is self._kanta_class:
|
if self._kanta_class is not None and bare is self._kanta_class:
|
||||||
@@ -516,7 +518,7 @@ class CallbackRegistry:
|
|||||||
@staticmethod
|
@staticmethod
|
||||||
def _unwrap_optional(ann: Any) -> Any:
|
def _unwrap_optional(ann: Any) -> Any:
|
||||||
origin = get_origin(ann)
|
origin = get_origin(ann)
|
||||||
if origin is not Union:
|
if origin not in (Union, types.UnionType):
|
||||||
return ann
|
return ann
|
||||||
args = [arg for arg in get_args(ann) if arg is not type(None)]
|
args = [arg for arg in get_args(ann) if arg is not type(None)]
|
||||||
return args[0] if len(args) == 1 else ann
|
return args[0] if len(args) == 1 else ann
|
||||||
@@ -524,7 +526,7 @@ class CallbackRegistry:
|
|||||||
@staticmethod
|
@staticmethod
|
||||||
def _is_optional_str(ann: Any) -> bool:
|
def _is_optional_str(ann: Any) -> bool:
|
||||||
origin = get_origin(ann)
|
origin = get_origin(ann)
|
||||||
if origin is not Union:
|
if origin not in (Union, types.UnionType):
|
||||||
return ann is str
|
return ann is str
|
||||||
args = get_args(ann)
|
args = get_args(ann)
|
||||||
return type(None) in args and any(arg is str for arg in args)
|
return type(None) in args and any(arg is str for arg in args)
|
||||||
|
|||||||
+28
-12
@@ -34,6 +34,7 @@ if sys.platform == "win32":
|
|||||||
_GENERIC_READ = 0x80000000
|
_GENERIC_READ = 0x80000000
|
||||||
_GENERIC_WRITE = 0x40000000
|
_GENERIC_WRITE = 0x40000000
|
||||||
_FILE_SHARE_READ = 0x00000001
|
_FILE_SHARE_READ = 0x00000001
|
||||||
|
_FILE_SHARE_WRITE = 0x00000002
|
||||||
_OPEN_EXISTING = 3
|
_OPEN_EXISTING = 3
|
||||||
_OPEN_ALWAYS = 4
|
_OPEN_ALWAYS = 4
|
||||||
_FILE_ATTRIBUTE_NORMAL = 0x80
|
_FILE_ATTRIBUTE_NORMAL = 0x80
|
||||||
@@ -91,12 +92,13 @@ else:
|
|||||||
|
|
||||||
|
|
||||||
class LockedFile:
|
class LockedFile:
|
||||||
"""A file opened with an exclusive write lock.
|
"""A file opened for read+write with an optional exclusive lock.
|
||||||
|
|
||||||
Usage::
|
Usage::
|
||||||
|
|
||||||
f = LockedFile()
|
f = LockedFile()
|
||||||
f.open(path) # open + lock (read+write)
|
f.open(path) # open + lock (read+write)
|
||||||
|
f.open(path, readonly=True) # open read-only without locking
|
||||||
content = f.read() # read entire content
|
content = f.read() # read entire content
|
||||||
f.write(data) # append data (seeks to end first)
|
f.write(data) # append data (seeks to end first)
|
||||||
f.close() # release lock + close fd
|
f.close() # release lock + close fd
|
||||||
@@ -108,12 +110,13 @@ class LockedFile:
|
|||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
self._fd: int | None = None # Unix fd or Windows HANDLE
|
self._fd: int | None = None # Unix fd or Windows HANDLE
|
||||||
|
|
||||||
def open(self, path: Path, *, create: bool = False) -> None:
|
def open(self, path: Path, *, create: bool = False, readonly: bool = False) -> None:
|
||||||
"""Open *path* for read+write with an exclusive lock.
|
"""Open *path* and optionally acquire an exclusive lock.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
path: File to open and lock.
|
path: File to open and lock.
|
||||||
create: If True, create the file if it doesn't exist (bootstrap).
|
create: If True, create the file if it doesn't exist (bootstrap).
|
||||||
|
readonly: If True, open read-only without acquiring a lock.
|
||||||
|
|
||||||
Raises:
|
Raises:
|
||||||
FileLockError: If the file is locked by another process or not found.
|
FileLockError: If the file is locked by another process or not found.
|
||||||
@@ -122,16 +125,18 @@ class LockedFile:
|
|||||||
return # Already open (idempotent)
|
return # Already open (idempotent)
|
||||||
|
|
||||||
if sys.platform == "win32":
|
if sys.platform == "win32":
|
||||||
self._open_win32(path, create)
|
self._open_win32(path, create, readonly)
|
||||||
else:
|
else:
|
||||||
self._open_unix(path, create)
|
self._open_unix(path, create, readonly)
|
||||||
|
|
||||||
def open_and_read(self, path: Path, create: bool = False) -> bytes:
|
def open_and_read(
|
||||||
"""Open *path* with exclusive lock and read all content.
|
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().
|
Combined operation for efficient use with asyncio.to_thread().
|
||||||
"""
|
"""
|
||||||
self.open(path, create=create)
|
self.open(path, create=create, readonly=readonly)
|
||||||
return self.read()
|
return self.read()
|
||||||
|
|
||||||
def read(self) -> bytes:
|
def read(self) -> bytes:
|
||||||
@@ -188,12 +193,16 @@ class LockedFile:
|
|||||||
|
|
||||||
# -- Unix ----------------------------------------------------------------
|
# -- Unix ----------------------------------------------------------------
|
||||||
|
|
||||||
def _open_unix(self, path: Path, create: bool) -> None:
|
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)
|
flags = os.O_RDWR | (os.O_CREAT if create else 0)
|
||||||
try:
|
try:
|
||||||
fd = os.open(path, flags, 0o666)
|
fd = os.open(path, flags, 0o666)
|
||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
_fatal(f"Database file not found: {path.resolve()}", db_path=path)
|
_fatal(f"Database file not found: {path.resolve()}", db_path=path)
|
||||||
|
if not readonly:
|
||||||
try:
|
try:
|
||||||
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||||
except OSError:
|
except OSError:
|
||||||
@@ -220,12 +229,19 @@ class LockedFile:
|
|||||||
|
|
||||||
# -- Windows -------------------------------------------------------------
|
# -- Windows -------------------------------------------------------------
|
||||||
|
|
||||||
def _open_win32(self, path: Path, create: bool) -> None:
|
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
|
disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING
|
||||||
|
access = _GENERIC_READ | _GENERIC_WRITE
|
||||||
|
share = _FILE_SHARE_READ
|
||||||
handle = _kernel32.CreateFileW(
|
handle = _kernel32.CreateFileW(
|
||||||
str(path),
|
str(path),
|
||||||
_GENERIC_READ | _GENERIC_WRITE,
|
access,
|
||||||
_FILE_SHARE_READ,
|
share,
|
||||||
None,
|
None,
|
||||||
disposition,
|
disposition,
|
||||||
_FILE_ATTRIBUTE_NORMAL,
|
_FILE_ATTRIBUTE_NORMAL,
|
||||||
|
|||||||
+54
-7
@@ -1,10 +1,11 @@
|
|||||||
"""Kanta DB main public API"""
|
"""Kanta DB main public API"""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
import logging
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from types import ModuleType
|
from types import ModuleType, SimpleNamespace
|
||||||
from typing import Any, Generic, TypeVar
|
from typing import Generic, TypeVar
|
||||||
|
|
||||||
from kanta.kantaimpl import KantaImpl
|
from kanta.kantaimpl import KantaImpl
|
||||||
from kanta.serialization import JsonSerializer, Serializer
|
from kanta.serialization import JsonSerializer, Serializer
|
||||||
@@ -50,7 +51,6 @@ class Kanta(Generic[T]):
|
|||||||
*,
|
*,
|
||||||
type: type[T] | None = None,
|
type: type[T] | None = None,
|
||||||
migrations: ModuleType | str | None = None,
|
migrations: ModuleType | str | None = None,
|
||||||
migration_ctx: Any | None = None,
|
|
||||||
serializer: Serializer | None = None,
|
serializer: Serializer | None = None,
|
||||||
flush_interval: float = 0.1,
|
flush_interval: float = 0.1,
|
||||||
):
|
):
|
||||||
@@ -61,7 +61,6 @@ class Kanta(Generic[T]):
|
|||||||
data: Caller-owned root msgspec.Struct state instance.
|
data: Caller-owned root msgspec.Struct state instance.
|
||||||
type: Optional explicit root type. Defaults to ``type(data)``.
|
type: Optional explicit root type. Defaults to ``type(data)``.
|
||||||
migrations: Optional migrations module object or import path.
|
migrations: Optional migrations module object or import path.
|
||||||
migration_ctx: Optional context object passed to migration functions.
|
|
||||||
flush_interval: Background flush interval in seconds.
|
flush_interval: Background flush interval in seconds.
|
||||||
serializer: Optional serializer implementation.
|
serializer: Optional serializer implementation.
|
||||||
|
|
||||||
@@ -78,7 +77,6 @@ class Kanta(Generic[T]):
|
|||||||
data=data,
|
data=data,
|
||||||
type=data_type,
|
type=data_type,
|
||||||
migrations=migrations,
|
migrations=migrations,
|
||||||
migration_ctx=migration_ctx,
|
|
||||||
flush_interval=flush_interval,
|
flush_interval=flush_interval,
|
||||||
kanta=self,
|
kanta=self,
|
||||||
)
|
)
|
||||||
@@ -127,6 +125,15 @@ class Kanta(Generic[T]):
|
|||||||
"""
|
"""
|
||||||
return self._impl.filename
|
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
|
@property
|
||||||
def mtime(self) -> datetime | None:
|
def mtime(self) -> datetime | None:
|
||||||
"""Last modification time carried forward from change records.
|
"""Last modification time carried forward from change records.
|
||||||
@@ -138,7 +145,13 @@ class Kanta(Generic[T]):
|
|||||||
"""
|
"""
|
||||||
return self._impl.mtime
|
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.
|
"""Open the database file and start background persistence.
|
||||||
|
|
||||||
This loads existing records, applies configured migrations, and starts
|
This loads existing records, applies configured migrations, and starts
|
||||||
@@ -147,6 +160,16 @@ class Kanta(Generic[T]):
|
|||||||
Args:
|
Args:
|
||||||
create: Whether to create the database file when missing.
|
create: Whether to create the database file when missing.
|
||||||
If False, opening fails when the file does not exist or is empty.
|
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 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.
|
Calling ``open`` more than once on the same instance is not allowed.
|
||||||
|
|
||||||
@@ -154,7 +177,7 @@ class Kanta(Generic[T]):
|
|||||||
kanta.exceptions.DatabaseError: If replay or decoding fails.
|
kanta.exceptions.DatabaseError: If replay or decoding fails.
|
||||||
kanta.exceptions.DataIntegrityError: If the instance is already open.
|
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]:
|
async def __aenter__(self) -> Kanta[T]:
|
||||||
"""Enter async context manager and open the database.
|
"""Enter async context manager and open the database.
|
||||||
@@ -230,6 +253,24 @@ class Kanta(Generic[T]):
|
|||||||
return _register
|
return _register
|
||||||
return _register(fn)
|
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):
|
def logfmt(self, fn=None, *, path: str | None = None):
|
||||||
"""Register a transaction logfmt callback.
|
"""Register a transaction logfmt callback.
|
||||||
|
|
||||||
@@ -257,6 +298,7 @@ class Kanta(Generic[T]):
|
|||||||
*,
|
*,
|
||||||
user: str | None = None,
|
user: str | None = None,
|
||||||
mtime: bool | datetime = True,
|
mtime: bool | datetime = True,
|
||||||
|
log: bool | logging.Logger = True,
|
||||||
):
|
):
|
||||||
"""Create a transactional mutation context manager.
|
"""Create a transactional mutation context manager.
|
||||||
|
|
||||||
@@ -271,6 +313,10 @@ class Kanta(Generic[T]):
|
|||||||
system operations that are not considered modifications. A
|
system operations that are not considered modifications. A
|
||||||
:class:`~datetime.datetime` value sets ``m`` to that explicit
|
:class:`~datetime.datetime` value sets ``m`` to that explicit
|
||||||
time.
|
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:
|
Returns:
|
||||||
A context manager yielding the live state object for mutation.
|
A context manager yielding the live state object for mutation.
|
||||||
@@ -285,4 +331,5 @@ class Kanta(Generic[T]):
|
|||||||
action,
|
action,
|
||||||
user=user,
|
user=user,
|
||||||
mtime=mtime,
|
mtime=mtime,
|
||||||
|
log=log,
|
||||||
)
|
)
|
||||||
|
|||||||
+171
-27
@@ -7,11 +7,13 @@ import copy
|
|||||||
import importlib
|
import importlib
|
||||||
import logging
|
import logging
|
||||||
from datetime import UTC, datetime
|
from datetime import UTC, datetime
|
||||||
|
from types import SimpleNamespace
|
||||||
from typing import Any, Generic, TypeVar
|
from typing import Any, Generic, TypeVar
|
||||||
|
|
||||||
from kanta.callbacks import CallbackRegistry, InjectionContext
|
from kanta.callbacks import CallbackRegistry, InjectionContext
|
||||||
from kanta.exceptions import DatabaseError, DataIntegrityError, ReplayError
|
from kanta.exceptions import DatabaseError, DataIntegrityError, ReplayError
|
||||||
from kanta.migrate import MigrationRegistry
|
from kanta.logging import _USER_PATH, bootstrap_logger, log_change, migration_logger
|
||||||
|
from kanta.migrations import MigrationResult, Migrations
|
||||||
from kanta.persistence import PersistenceMixin
|
from kanta.persistence import PersistenceMixin
|
||||||
from kanta.serialization import restore_data_in_place, struct_to_dict
|
from kanta.serialization import restore_data_in_place, struct_to_dict
|
||||||
from kanta.serialization.base import replay
|
from kanta.serialization.base import replay
|
||||||
@@ -27,22 +29,23 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
|||||||
def __init__(self, **kwargs: Any):
|
def __init__(self, **kwargs: Any):
|
||||||
self.data_type = kwargs.pop("type")
|
self.data_type = kwargs.pop("type")
|
||||||
self.data: T = kwargs.pop("data")
|
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)
|
self._kanta = kwargs.pop("kanta", None)
|
||||||
|
migrations = kwargs.pop("migrations", None)
|
||||||
|
self.ctx = SimpleNamespace()
|
||||||
super().__init__(**kwargs)
|
super().__init__(**kwargs)
|
||||||
self.migration_registry: MigrationRegistry | None = None
|
self.migrations: Migrations | None = None
|
||||||
if self.migrations is not None:
|
if migrations is not None:
|
||||||
module = (
|
module = (
|
||||||
importlib.import_module(self.migrations)
|
importlib.import_module(migrations)
|
||||||
if isinstance(self.migrations, str)
|
if isinstance(migrations, str)
|
||||||
else self.migrations
|
else migrations
|
||||||
)
|
)
|
||||||
self.migration_registry = MigrationRegistry.from_module(module)
|
self.migrations = Migrations.from_module(module)
|
||||||
|
|
||||||
self.in_transaction = False
|
self.in_transaction = False
|
||||||
self.transaction_snapshot: dict[str, Any] | None = None
|
self.transaction_snapshot: dict[str, Any] | None = None
|
||||||
self.opened = False
|
self.opened = False
|
||||||
|
self.readonly = False
|
||||||
self.bootstrap_action = "bootstrap"
|
self.bootstrap_action = "bootstrap"
|
||||||
self.bootstrap_user: str | None = None
|
self.bootstrap_user: str | None = None
|
||||||
self.bootstrap_mtime: bool | datetime = True
|
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.statedict = struct_to_dict(self.data, serializer=self.serializer)
|
||||||
self.version = (
|
self.version = self.migrations.dbver if self.migrations is not None else 0
|
||||||
self.migration_registry.dbver if self.migration_registry is not None else 0
|
|
||||||
)
|
|
||||||
|
|
||||||
def add_bootstrap(
|
def add_bootstrap(
|
||||||
self,
|
self,
|
||||||
@@ -75,7 +76,64 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
|||||||
"""Register one transaction logfmt callback."""
|
"""Register one transaction logfmt callback."""
|
||||||
self.callback_registry.register("logfmt", callback, path=path)
|
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."""
|
"""Open the database: load from disk, apply migrations, start background task."""
|
||||||
if self.opened:
|
if self.opened:
|
||||||
raise DataIntegrityError(
|
raise DataIntegrityError(
|
||||||
@@ -84,12 +142,17 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
|||||||
action="open",
|
action="open",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
self.readonly = readonly
|
||||||
existed_before_open = self.filename.exists()
|
existed_before_open = self.filename.exists()
|
||||||
|
|
||||||
|
# Read-only mode never creates the file.
|
||||||
|
open_create = create and not readonly
|
||||||
|
|
||||||
content = await asyncio.to_thread(
|
content = await asyncio.to_thread(
|
||||||
self.file.open_and_read,
|
self.file.open_and_read,
|
||||||
self.filename,
|
self.filename,
|
||||||
create=create,
|
create=open_create,
|
||||||
|
readonly=readonly,
|
||||||
)
|
)
|
||||||
|
|
||||||
if not create and (not existed_before_open or not content):
|
if not create and (not existed_before_open or not content):
|
||||||
@@ -105,6 +168,9 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
|||||||
action="open",
|
action="open",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# From this point the file is open and must be closed via close().
|
||||||
|
self.opened = True
|
||||||
|
|
||||||
if content:
|
if content:
|
||||||
try:
|
try:
|
||||||
rr = replay(
|
rr = replay(
|
||||||
@@ -134,12 +200,33 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
|||||||
cause_type=type(e).__name__,
|
cause_type=type(e).__name__,
|
||||||
) from e
|
) from e
|
||||||
|
|
||||||
if self.migration_registry is not None:
|
migration_result = None
|
||||||
rr.version = self.migration_registry.apply(
|
state_before_migrations = None
|
||||||
rr.state, rr.version, self.migration_ctx
|
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
|
||||||
|
|
||||||
|
migrations_ran = rr.version != previous_version
|
||||||
|
migration_state_changed = (
|
||||||
|
state_before_migrations is not None
|
||||||
|
and state_before_migrations != rr.state
|
||||||
)
|
)
|
||||||
|
|
||||||
self.statedict = copy.deepcopy(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 = restore_data_in_place(
|
||||||
self.data,
|
self.data,
|
||||||
rr.state,
|
rr.state,
|
||||||
@@ -148,28 +235,85 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
|||||||
)
|
)
|
||||||
self.version = rr.version
|
self.version = rr.version
|
||||||
self.mtime = rr.m
|
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)
|
normalized = struct_to_dict(self.data, serializer=self.serializer)
|
||||||
self.queue_change("migrate:msgspec", normalized, mtime=False)
|
if self.readonly:
|
||||||
self.snapshot.ts = (
|
self.statedict = copy.deepcopy(normalized)
|
||||||
datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)
|
else:
|
||||||
if rr.last_snapshot_mtime is not None
|
if migrations_ran and migration_state_changed:
|
||||||
else None
|
self.queue_change(
|
||||||
|
f"migrate:v{self.version}",
|
||||||
|
rr.state,
|
||||||
|
mtime=False,
|
||||||
)
|
)
|
||||||
elif self.callback_registry.has("bootstrap"):
|
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:
|
try:
|
||||||
|
if self.callback_registry.has("bootstrap"):
|
||||||
await self.callback_registry.invoke(
|
await self.callback_registry.invoke(
|
||||||
"bootstrap",
|
"bootstrap",
|
||||||
InjectionContext(data=self.data, kanta=self._kanta),
|
InjectionContext(data=self.data, kanta=self._kanta),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
self.statedict = {}
|
||||||
current = struct_to_dict(self.data, serializer=self.serializer)
|
current = struct_to_dict(self.data, serializer=self.serializer)
|
||||||
self.queue_change(
|
record = self.queue_change(
|
||||||
self.bootstrap_action,
|
self.bootstrap_action,
|
||||||
current,
|
current,
|
||||||
user=self.bootstrap_user,
|
user=self.bootstrap_user,
|
||||||
mtime=self.bootstrap_mtime,
|
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:
|
except Exception:
|
||||||
|
self.opened = False
|
||||||
self.file.close()
|
self.file.close()
|
||||||
try:
|
try:
|
||||||
await asyncio.to_thread(self.filename.unlink, missing_ok=True)
|
await asyncio.to_thread(self.filename.unlink, missing_ok=True)
|
||||||
@@ -177,8 +321,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
|||||||
pass
|
pass
|
||||||
raise
|
raise
|
||||||
|
|
||||||
self.opened = True
|
if not self.readonly:
|
||||||
|
|
||||||
self.background_task = asyncio.create_task(self._background_loop())
|
self.background_task = asyncio.create_task(self._background_loop())
|
||||||
|
|
||||||
async def close(self) -> None:
|
async def close(self) -> None:
|
||||||
@@ -196,6 +339,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
|||||||
|
|
||||||
# Always run a final flush in case the background task never reached
|
# Always run a final flush in case the background task never reached
|
||||||
# its cancellation handler.
|
# its cancellation handler.
|
||||||
|
if not self.readonly:
|
||||||
await self.flush()
|
await self.flush()
|
||||||
|
|
||||||
self.file.close()
|
self.file.close()
|
||||||
|
|||||||
+55
-13
@@ -1,7 +1,8 @@
|
|||||||
"""Database change logging with pretty-printed diffs.
|
"""Database change logging with pretty-printed diffs.
|
||||||
|
|
||||||
Provides a logger for JSONL database changes that formats diffs
|
Provides loggers for JSONL database changes, bootstrap events, and
|
||||||
in a human-readable path.notation style with color coding.
|
migrations. Diff output is formatted in a human-readable path notation
|
||||||
|
style with color coding.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
@@ -10,7 +11,9 @@ import sys
|
|||||||
from collections.abc import Callable
|
from collections.abc import Callable
|
||||||
from typing import Any
|
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
|
# Pattern to match control characters and bidirectional overrides
|
||||||
_UNSAFE_CHARS = re.compile(
|
_UNSAFE_CHARS = re.compile(
|
||||||
@@ -274,6 +277,9 @@ def log_change(
|
|||||||
user: str | None = None,
|
user: str | None = None,
|
||||||
previous: dict | None = None,
|
previous: dict | None = None,
|
||||||
logfmt: Callable[[Any, str], str | None] | None = None,
|
logfmt: Callable[[Any, str], str | None] | None = None,
|
||||||
|
*,
|
||||||
|
logger: logging.Logger = transaction_logger,
|
||||||
|
level: int = logging.INFO,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Log a database change with pretty-printed diff.
|
"""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.
|
user: Optional already-formatted user name to show in the header.
|
||||||
previous: The previous state dict (for determining add vs update).
|
previous: The previous state dict (for determining add vs update).
|
||||||
logfmt: Optional formatter callable ``(value, path) -> str | None``.
|
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)
|
header = format_action_header(action, user)
|
||||||
diff_lines = format_diff(diff, previous, logfmt)
|
diff_lines = format_diff(diff, previous, logfmt)
|
||||||
|
|
||||||
if not diff_lines:
|
if not diff_lines:
|
||||||
logger.info(header)
|
logger.log(level, header)
|
||||||
return
|
return
|
||||||
|
|
||||||
if len(diff_lines) == 1:
|
if len(diff_lines) == 1:
|
||||||
logger.info(f"{header}{diff_lines[0]}")
|
logger.log(level, f"{header}{diff_lines[0]}")
|
||||||
else:
|
else:
|
||||||
logger.info(header)
|
logger.log(level, header)
|
||||||
for line in diff_lines:
|
for line in diff_lines:
|
||||||
logger.info(line)
|
logger.log(level, line)
|
||||||
|
|
||||||
|
|
||||||
def configure_logging() -> None:
|
def configure_logging(
|
||||||
"""Configure the database logger to output to stderr without prefix."""
|
*,
|
||||||
if not logger.handlers:
|
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 = logging.StreamHandler(sys.stderr)
|
||||||
handler.setFormatter(logging.Formatter("%(message)s"))
|
handler.setFormatter(logging.Formatter("%(message)s"))
|
||||||
logger.addHandler(handler)
|
target.addHandler(handler)
|
||||||
logger.setLevel(logging.INFO)
|
target.setLevel(logging.INFO)
|
||||||
logger.propagate = False
|
|
||||||
|
|||||||
@@ -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
|
|
||||||
@@ -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)
|
||||||
+23
-1
@@ -39,6 +39,7 @@ class PersistenceMixin:
|
|||||||
flush_interval: float
|
flush_interval: float
|
||||||
version: int
|
version: int
|
||||||
opened: bool
|
opened: bool
|
||||||
|
readonly: bool
|
||||||
mtime: datetime | None
|
mtime: datetime | None
|
||||||
|
|
||||||
def __init__(self, **kwargs: Any) -> None:
|
def __init__(self, **kwargs: Any) -> None:
|
||||||
@@ -68,6 +69,8 @@ class PersistenceMixin:
|
|||||||
|
|
||||||
async def _background_loop(self) -> None:
|
async def _background_loop(self) -> None:
|
||||||
"""Background task that periodically flushes changes to disk."""
|
"""Background task that periodically flushes changes to disk."""
|
||||||
|
if self.readonly:
|
||||||
|
return
|
||||||
while True:
|
while True:
|
||||||
try:
|
try:
|
||||||
await asyncio.sleep(self.flush_interval)
|
await asyncio.sleep(self.flush_interval)
|
||||||
@@ -106,6 +109,7 @@ class PersistenceMixin:
|
|||||||
*,
|
*,
|
||||||
user: str | None = None,
|
user: str | None = None,
|
||||||
mtime: bool | datetime = True,
|
mtime: bool | datetime = True,
|
||||||
|
force: bool = False,
|
||||||
) -> ChangeRecord | None:
|
) -> ChangeRecord | None:
|
||||||
"""Queue a change record internally (thread-safe).
|
"""Queue a change record internally (thread-safe).
|
||||||
|
|
||||||
@@ -118,9 +122,11 @@ class PersistenceMixin:
|
|||||||
previous modification time remains in effect; this is used for
|
previous modification time remains in effect; this is used for
|
||||||
system operations that are not considered modifications. A
|
system operations that are not considered modifications. A
|
||||||
:class:`~datetime.datetime` value sets ``m`` to that explicit time.
|
:class:`~datetime.datetime` value sets ``m`` to that explicit time.
|
||||||
|
force: If ``True``, queue the record even when the diff is empty.
|
||||||
|
|
||||||
Returns:
|
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)
|
now = datetime.now(UTC)
|
||||||
|
|
||||||
@@ -135,7 +141,9 @@ class PersistenceMixin:
|
|||||||
|
|
||||||
diff = compute_diff(self.statedict, current)
|
diff = compute_diff(self.statedict, current)
|
||||||
if not diff:
|
if not diff:
|
||||||
|
if not force:
|
||||||
return None
|
return None
|
||||||
|
diff = {}
|
||||||
|
|
||||||
record = ChangeRecord(
|
record = ChangeRecord(
|
||||||
ts=now,
|
ts=now,
|
||||||
@@ -160,6 +168,13 @@ class PersistenceMixin:
|
|||||||
action="flush_sync",
|
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:
|
if self.flush_failed:
|
||||||
return
|
return
|
||||||
|
|
||||||
@@ -207,6 +222,13 @@ class PersistenceMixin:
|
|||||||
action="flush",
|
action="flush",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
if self.readonly:
|
||||||
|
raise DataIntegrityError(
|
||||||
|
"Cannot flush in read-only mode",
|
||||||
|
db_path=self.filename,
|
||||||
|
action="flush",
|
||||||
|
)
|
||||||
|
|
||||||
if self.flush_failed:
|
if self.flush_failed:
|
||||||
return
|
return
|
||||||
|
|
||||||
|
|||||||
+5
-4
@@ -41,14 +41,15 @@ class SnapshotState:
|
|||||||
self, file, version: int, state: dict, m: datetime | None = None
|
self, file, version: int, state: dict, m: datetime | None = None
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Write snapshot when thresholds/time policy allows it."""
|
"""Write snapshot when thresholds/time policy allows it."""
|
||||||
if self.changes < self._min_diffs:
|
|
||||||
return
|
|
||||||
force = self._force_pending
|
force = self._force_pending
|
||||||
now = datetime.now(UTC)
|
now = datetime.now(UTC)
|
||||||
if not force and now.weekday() != 6: # 6 = Sunday
|
if not force:
|
||||||
|
if self.changes < self._min_diffs:
|
||||||
|
return
|
||||||
|
if now.weekday() != 6: # 6 = Sunday
|
||||||
return
|
return
|
||||||
sunday_midnight = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
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:
|
if self.ts is not None and self.ts >= sunday_midnight:
|
||||||
return
|
return
|
||||||
if not file.is_open:
|
if not file.is_open:
|
||||||
return
|
return
|
||||||
|
|||||||
+1
-1
@@ -23,7 +23,7 @@ class ChangeRecord(msgspec.Struct, omit_defaults=True, kw_only=True):
|
|||||||
v: int = 0
|
v: int = 0
|
||||||
u: str | None = None
|
u: str | None = None
|
||||||
m: datetime | None = None
|
m: datetime | None = None
|
||||||
diff: dict
|
diff: dict = {}
|
||||||
|
|
||||||
|
|
||||||
class Snapshot(msgspec.Struct, omit_defaults=True):
|
class Snapshot(msgspec.Struct, omit_defaults=True):
|
||||||
|
|||||||
+19
-2
@@ -9,7 +9,7 @@ from datetime import datetime
|
|||||||
from kanta.diff import compute_diff
|
from kanta.diff import compute_diff
|
||||||
from kanta.exceptions import DataIntegrityError
|
from kanta.exceptions import DataIntegrityError
|
||||||
from kanta.callbacks import InjectionContext
|
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
|
from kanta.serialization import restore_data_in_place, struct_to_dict
|
||||||
|
|
||||||
_logger = logging.getLogger(__name__)
|
_logger = logging.getLogger(__name__)
|
||||||
@@ -22,8 +22,16 @@ def transaction(
|
|||||||
*,
|
*,
|
||||||
user: str | None = None,
|
user: str | None = None,
|
||||||
mtime: bool | datetime = True,
|
mtime: bool | datetime = True,
|
||||||
|
log: bool | logging.Logger = True,
|
||||||
):
|
):
|
||||||
"""Wrap writes in a transaction and yield the live db object."""
|
"""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:
|
if impl.in_transaction:
|
||||||
raise RuntimeError(
|
raise RuntimeError(
|
||||||
"Nested or simultaneous transactions are not supported "
|
"Nested or simultaneous transactions are not supported "
|
||||||
@@ -73,7 +81,16 @@ def transaction(
|
|||||||
resolved = logfmt(user, _USER_PATH)
|
resolved = logfmt(user, _USER_PATH)
|
||||||
if resolved is not None:
|
if resolved is not None:
|
||||||
formatted_user = resolved
|
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:
|
except Exception:
|
||||||
_logger.warning("Transaction '%s' failed, rolling back changes", action)
|
_logger.warning("Transaction '%s' failed, rolling back changes", action)
|
||||||
if impl.transaction_snapshot is not None:
|
if impl.transaction_snapshot is not None:
|
||||||
|
|||||||
+24
-1
@@ -7,7 +7,7 @@ from uuid import UUID
|
|||||||
import msgspec
|
import msgspec
|
||||||
|
|
||||||
from kanta.kanta import Kanta
|
from kanta.kanta import Kanta
|
||||||
from kanta.structs import ChangeRecord
|
from kanta.structs import ChangeRecord, Snapshot
|
||||||
|
|
||||||
|
|
||||||
class User(msgspec.Struct):
|
class User(msgspec.Struct):
|
||||||
@@ -70,6 +70,18 @@ def change_actions(path: Path, format_config) -> list[str]:
|
|||||||
return actions
|
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):
|
def make_migrations_module(name: str, fn_name: str, fn):
|
||||||
mod = ModuleType(name)
|
mod = ModuleType(name)
|
||||||
mod.__dict__[fn_name] = fn
|
mod.__dict__[fn_name] = fn
|
||||||
@@ -77,6 +89,17 @@ def make_migrations_module(name: str, fn_name: str, fn):
|
|||||||
return mod
|
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:
|
def fixed_change(action: str, diff: dict, *, version: int = 0) -> ChangeRecord:
|
||||||
return ChangeRecord(
|
return ChangeRecord(
|
||||||
ts=datetime(2026, 1, 1, tzinfo=UTC), a=action, v=version, diff=diff
|
ts=datetime(2026, 1, 1, tzinfo=UTC), a=action, v=version, diff=diff
|
||||||
|
|||||||
+57
-11
@@ -1,4 +1,4 @@
|
|||||||
from typing import Any
|
from typing import Any, Optional, Union
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
@@ -49,16 +49,60 @@ def test_logfmt_requires_value_annotation(tmp_path, format_config):
|
|||||||
return None
|
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)
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
||||||
|
|
||||||
with pytest.raises(TypeError, match="must annotate its return"):
|
|
||||||
|
|
||||||
@kanta.logfmt
|
@kanta.logfmt
|
||||||
def resolve_names(value: str, current: DictPost):
|
def resolve_names(value: str, current: DictPost):
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
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):
|
def test_logfmt_rejects_async_callback(tmp_path, format_config):
|
||||||
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
||||||
|
|
||||||
@@ -104,7 +148,7 @@ async def test_bootstrap_injects_kanta(tmp_path, format_config):
|
|||||||
async def test_logfmt_injects_states(tmp_path, format_config, caplog):
|
async def test_logfmt_injects_states(tmp_path, format_config, caplog):
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
caplog.set_level(logging.INFO, logger="kanta.changes")
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
||||||
path = tmp_path / "test.db"
|
path = tmp_path / "test.db"
|
||||||
kanta = make_kanta(path, Data, format_config)
|
kanta = make_kanta(path, Data, format_config)
|
||||||
|
|
||||||
@@ -126,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):
|
async def test_logfmt_class_injection(tmp_path, format_config, caplog):
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
caplog.set_level(logging.INFO, logger="kanta.changes")
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
||||||
path = tmp_path / "test.db"
|
path = tmp_path / "test.db"
|
||||||
kanta = make_kanta(path, Data, format_config)
|
kanta = make_kanta(path, Data, format_config)
|
||||||
|
|
||||||
@kanta.logfmt
|
@kanta.logfmt
|
||||||
class UserLogFmt(LogFmt):
|
class UserLogFmt(LogFmt):
|
||||||
def resolve(self, value: str, path: str) -> str | None:
|
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")
|
return self.current_state.get("users", {}).get(value, {}).get("name")
|
||||||
|
|
||||||
await kanta.open()
|
await kanta.open()
|
||||||
@@ -149,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):
|
async def test_multiple_logfmt_chain(tmp_path, format_config, caplog):
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
caplog.set_level(logging.INFO, logger="kanta.changes")
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
||||||
path = tmp_path / "test.db"
|
path = tmp_path / "test.db"
|
||||||
kanta = make_kanta(path, Data, format_config)
|
kanta = make_kanta(path, Data, format_config)
|
||||||
|
|
||||||
@@ -177,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):
|
async def test_logfmt_path_context(tmp_path, format_config, caplog):
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
caplog.set_level(logging.INFO, logger="kanta.changes")
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
||||||
path = tmp_path / "test.db"
|
path = tmp_path / "test.db"
|
||||||
kanta = make_kanta(path, Data, format_config)
|
kanta = make_kanta(path, Data, format_config)
|
||||||
|
|
||||||
@@ -201,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):
|
async def test_logfmt_decorator_path_filters_calls(tmp_path, format_config, caplog):
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
caplog.set_level(logging.INFO, logger="kanta.changes")
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
||||||
path = tmp_path / "test.db"
|
path = tmp_path / "test.db"
|
||||||
kanta = make_kanta(path, Data, format_config)
|
kanta = make_kanta(path, Data, format_config)
|
||||||
|
|
||||||
@@ -227,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):
|
async def test_logfmt_user_path_replaces_user_display(tmp_path, format_config, caplog):
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
caplog.set_level(logging.INFO, logger="kanta.changes")
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
||||||
path = tmp_path / "test.db"
|
path = tmp_path / "test.db"
|
||||||
kanta = make_kanta(path, Data, format_config)
|
kanta = make_kanta(path, Data, format_config)
|
||||||
|
|
||||||
@@ -249,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):
|
async def test_logfmt_non_string_value(tmp_path, format_config, caplog):
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
caplog.set_level(logging.INFO, logger="kanta.changes")
|
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
||||||
path = tmp_path / "test.db"
|
path = tmp_path / "test.db"
|
||||||
kanta = make_kanta(path, Data, format_config)
|
kanta = make_kanta(path, Data, format_config)
|
||||||
|
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
import asyncio
|
import asyncio
|
||||||
|
import logging
|
||||||
import sys
|
import sys
|
||||||
from datetime import UTC, datetime
|
from datetime import UTC, datetime
|
||||||
from uuid import uuid4
|
from uuid import uuid4
|
||||||
@@ -6,6 +7,7 @@ from uuid import uuid4
|
|||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from kanta.exceptions import DatabaseError, DataIntegrityError, FileLockError
|
from kanta.exceptions import DatabaseError, DataIntegrityError, FileLockError
|
||||||
|
from kanta.migrations import MigrationResult
|
||||||
from kanta.serialization import struct_to_dict
|
from kanta.serialization import struct_to_dict
|
||||||
|
|
||||||
from .support import (
|
from .support import (
|
||||||
@@ -17,6 +19,9 @@ from .support import (
|
|||||||
change_actions,
|
change_actions,
|
||||||
fixed_change,
|
fixed_change,
|
||||||
make_kanta,
|
make_kanta,
|
||||||
|
make_migrations_module,
|
||||||
|
read_changes,
|
||||||
|
read_last_snapshot,
|
||||||
seed_single_change,
|
seed_single_change,
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -30,6 +35,63 @@ async def test_load_empty(tmp_path, format_config):
|
|||||||
await kanta.close()
|
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
|
@pytest.mark.asyncio
|
||||||
async def test_open_overwrites_caller_owned_root_data(tmp_path, format_config):
|
async def test_open_overwrites_caller_owned_root_data(tmp_path, format_config):
|
||||||
path = tmp_path / "test.db"
|
path = tmp_path / "test.db"
|
||||||
@@ -404,7 +466,7 @@ async def test_migrations_from_module(tmp_path, format_config):
|
|||||||
|
|
||||||
mod = type(sys)("test_migrations")
|
mod = type(sys)("test_migrations")
|
||||||
|
|
||||||
def migrate_v1(d, ctx):
|
def migrate_v1(d, kanta):
|
||||||
d["version"] = 1
|
d["version"] = 1
|
||||||
|
|
||||||
mod.__dict__["migrate_v1"] = migrate_v1
|
mod.__dict__["migrate_v1"] = migrate_v1
|
||||||
@@ -434,6 +496,268 @@ async def test_msgspec_normalization_logs_migration(tmp_path, format_config):
|
|||||||
assert "migrate:msgspec" in change_actions(path, format_config)
|
assert "migrate:msgspec" in change_actions(path, format_config)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_empty_migration_writes_snapshot_and_is_not_reapplied(
|
||||||
|
tmp_path, format_config
|
||||||
|
):
|
||||||
|
path = tmp_path / "test.db"
|
||||||
|
seed_single_change(
|
||||||
|
path, fixed_change("init", {"counter": 0, "users": {}}), format_config
|
||||||
|
)
|
||||||
|
|
||||||
|
def migrate_v1(d, kanta):
|
||||||
|
"""No-op migration that only bumps the schema version."""
|
||||||
|
pass
|
||||||
|
|
||||||
|
mod = make_migrations_module("empty_migration_mod", "migrate_v1", migrate_v1)
|
||||||
|
|
||||||
|
try:
|
||||||
|
kanta = make_kanta(path, Data, format_config, migrations=mod)
|
||||||
|
await kanta.open()
|
||||||
|
assert kanta.version == 1
|
||||||
|
await kanta.close()
|
||||||
|
|
||||||
|
# Empty migrations must not produce empty change records.
|
||||||
|
records = read_changes(path, format_config)
|
||||||
|
migration_records = [r for r in records if r.a.startswith("migrate")]
|
||||||
|
assert not migration_records
|
||||||
|
|
||||||
|
# The version bump is persisted via a snapshot instead.
|
||||||
|
snap = read_last_snapshot(path, format_config)
|
||||||
|
assert snap is not None
|
||||||
|
assert snap.v == 1
|
||||||
|
assert snap.state == {"counter": 0, "users": {}}
|
||||||
|
|
||||||
|
kanta2 = make_kanta(path, Data, format_config, migrations=mod)
|
||||||
|
await kanta2.open()
|
||||||
|
assert kanta2.version == 1
|
||||||
|
await kanta2.close()
|
||||||
|
|
||||||
|
# Re-opening must not create additional migration records or snapshots.
|
||||||
|
records2 = read_changes(path, format_config)
|
||||||
|
assert not [r for r in records2 if r.a.startswith("migrate")]
|
||||||
|
finally:
|
||||||
|
sys.modules.pop("empty_migration_mod", None)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_migration_with_changes_records_diff_and_snapshot(
|
||||||
|
tmp_path, format_config
|
||||||
|
):
|
||||||
|
path = tmp_path / "test.db"
|
||||||
|
seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
|
||||||
|
|
||||||
|
mod = type(sys)("test_migrations_changes")
|
||||||
|
|
||||||
|
def migrate_v1(d, kanta):
|
||||||
|
d["counter"] = 2
|
||||||
|
|
||||||
|
mod.__dict__["migrate_v1"] = migrate_v1
|
||||||
|
|
||||||
|
kanta = make_kanta(path, Data, format_config, migrations=mod)
|
||||||
|
await kanta.open()
|
||||||
|
assert kanta.version == 1
|
||||||
|
assert kanta.data.counter == 2
|
||||||
|
await kanta.close()
|
||||||
|
|
||||||
|
records = read_changes(path, format_config)
|
||||||
|
migration_records = [r for r in records if r.a.startswith("migrate")]
|
||||||
|
assert len(migration_records) == 2
|
||||||
|
assert migration_records[0].a == "migrate:v1"
|
||||||
|
assert migration_records[0].v == 1
|
||||||
|
assert migration_records[0].diff == {"counter": 2}
|
||||||
|
assert migration_records[1].a == "migrate:msgspec"
|
||||||
|
assert migration_records[1].v == 1
|
||||||
|
assert migration_records[1].diff == {"users": {}}
|
||||||
|
|
||||||
|
snap = read_last_snapshot(path, format_config)
|
||||||
|
assert snap is not None
|
||||||
|
assert snap.v == 1
|
||||||
|
assert snap.state == {"counter": 2, "users": {}}
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_migration_summary_log_includes_filename(tmp_path, format_config, caplog):
|
||||||
|
path = tmp_path / "test.db"
|
||||||
|
seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
|
||||||
|
|
||||||
|
mod = type(sys)("test_migrations_log")
|
||||||
|
|
||||||
|
def migrate_v1(d, kanta):
|
||||||
|
"""Bump counter."""
|
||||||
|
d["counter"] = 2
|
||||||
|
|
||||||
|
mod.__dict__["migrate_v1"] = migrate_v1
|
||||||
|
|
||||||
|
with caplog.at_level(logging.INFO, logger="kanta.migration"):
|
||||||
|
kanta = make_kanta(path, Data, format_config, migrations=mod)
|
||||||
|
await kanta.open()
|
||||||
|
assert kanta.version == 1
|
||||||
|
await kanta.close()
|
||||||
|
|
||||||
|
info_messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
|
||||||
|
assert len(info_messages) == 1
|
||||||
|
assert str(path) in info_messages[0]
|
||||||
|
assert "v0 -> v1" in info_messages[0]
|
||||||
|
assert "migrate_v1 (Bump counter)" in info_messages[0]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_open_log_false_suppresses_migration_log(tmp_path, format_config, caplog):
|
||||||
|
path = tmp_path / "test.db"
|
||||||
|
seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
|
||||||
|
|
||||||
|
mod = type(sys)("test_migrations_silent")
|
||||||
|
|
||||||
|
def migrate_v1(d, kanta):
|
||||||
|
d["counter"] = 2
|
||||||
|
|
||||||
|
mod.__dict__["migrate_v1"] = migrate_v1
|
||||||
|
|
||||||
|
with caplog.at_level(logging.INFO, logger="kanta.migration"):
|
||||||
|
kanta = make_kanta(path, Data, format_config, migrations=mod)
|
||||||
|
await kanta.open(log=False)
|
||||||
|
await kanta.close()
|
||||||
|
|
||||||
|
info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
|
||||||
|
assert not info_messages
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_open_log_true_logs_bootstrap(tmp_path, format_config, caplog):
|
||||||
|
path = tmp_path / "test.db"
|
||||||
|
kanta = make_kanta(path, Data, format_config)
|
||||||
|
|
||||||
|
with caplog.at_level(logging.INFO, logger="kanta.bootstrap"):
|
||||||
|
await kanta.open()
|
||||||
|
await kanta.close()
|
||||||
|
|
||||||
|
info_messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
|
||||||
|
assert len(info_messages) >= 2
|
||||||
|
assert "Created" in info_messages[0]
|
||||||
|
assert "bootstrap" in info_messages[1]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_open_log_false_suppresses_bootstrap_log(tmp_path, format_config, caplog):
|
||||||
|
path = tmp_path / "test.db"
|
||||||
|
kanta = make_kanta(path, Data, format_config)
|
||||||
|
|
||||||
|
with caplog.at_level(logging.INFO, logger="kanta.bootstrap"):
|
||||||
|
await kanta.open(log=False)
|
||||||
|
await kanta.close()
|
||||||
|
|
||||||
|
info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
|
||||||
|
assert not info_messages
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_open_log_custom_logger_logs_bootstrap(tmp_path, format_config, caplog):
|
||||||
|
path = tmp_path / "test.db"
|
||||||
|
kanta = make_kanta(path, Data, format_config)
|
||||||
|
|
||||||
|
custom_logger = logging.getLogger("custom.bootstrap")
|
||||||
|
custom_logger.setLevel(logging.INFO)
|
||||||
|
|
||||||
|
with caplog.at_level(logging.INFO, logger="custom.bootstrap"):
|
||||||
|
await kanta.open(log=custom_logger)
|
||||||
|
await kanta.close()
|
||||||
|
|
||||||
|
info_messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
|
||||||
|
assert len(info_messages) >= 2
|
||||||
|
assert "Created" in info_messages[0]
|
||||||
|
assert "bootstrap" in info_messages[1]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_open_existing_database_logs_using_on_debug(
|
||||||
|
tmp_path, format_config, caplog
|
||||||
|
):
|
||||||
|
path = tmp_path / "test.db"
|
||||||
|
kanta = make_kanta(path, Data, format_config)
|
||||||
|
await kanta.open()
|
||||||
|
await kanta.close()
|
||||||
|
|
||||||
|
kanta2 = make_kanta(path, Data, format_config)
|
||||||
|
|
||||||
|
with caplog.at_level(logging.DEBUG, logger="kanta.bootstrap"):
|
||||||
|
await kanta2.open()
|
||||||
|
await kanta2.close()
|
||||||
|
|
||||||
|
debug_messages = [r.message for r in caplog.records if r.levelno == logging.DEBUG]
|
||||||
|
assert any("Using" in m and str(path.resolve()) in m for m in debug_messages)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_logmigr_callback_replaces_default_logging(
|
||||||
|
tmp_path, format_config, caplog
|
||||||
|
):
|
||||||
|
path = tmp_path / "test.db"
|
||||||
|
seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
|
||||||
|
|
||||||
|
mod = type(sys)("test_migrations_callback")
|
||||||
|
|
||||||
|
def migrate_v1(d, kanta):
|
||||||
|
"""Bump counter."""
|
||||||
|
d["counter"] = 2
|
||||||
|
|
||||||
|
mod.__dict__["migrate_v1"] = migrate_v1
|
||||||
|
|
||||||
|
summaries = []
|
||||||
|
|
||||||
|
kanta = make_kanta(path, Data, format_config, migrations=mod)
|
||||||
|
|
||||||
|
@kanta.logmigr
|
||||||
|
def collect(summary: MigrationResult):
|
||||||
|
summaries.append(summary)
|
||||||
|
|
||||||
|
with caplog.at_level(logging.INFO, logger="kanta.migration"):
|
||||||
|
await kanta.open()
|
||||||
|
await kanta.close()
|
||||||
|
|
||||||
|
assert len(summaries) == 1
|
||||||
|
assert summaries[0].version == 1
|
||||||
|
assert summaries[0].migrations[0].name == "migrate_v1"
|
||||||
|
info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
|
||||||
|
assert not info_messages
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_transaction_log_false_suppresses_log(tmp_path, format_config, caplog):
|
||||||
|
path = tmp_path / "test.db"
|
||||||
|
kanta = make_kanta(path, Data, format_config)
|
||||||
|
await kanta.open()
|
||||||
|
|
||||||
|
with caplog.at_level(logging.INFO, logger="kanta.transaction"):
|
||||||
|
with kanta.transaction(action="inc", log=False) as data:
|
||||||
|
data.counter = 1
|
||||||
|
|
||||||
|
await kanta.close()
|
||||||
|
|
||||||
|
info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
|
||||||
|
assert not info_messages
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_transaction_log_custom_logger(tmp_path, format_config, caplog):
|
||||||
|
path = tmp_path / "test.db"
|
||||||
|
kanta = make_kanta(path, Data, format_config)
|
||||||
|
await kanta.open()
|
||||||
|
|
||||||
|
custom_logger = logging.getLogger("custom.transaction")
|
||||||
|
custom_logger.setLevel(logging.INFO)
|
||||||
|
|
||||||
|
with caplog.at_level(logging.INFO, logger="custom.transaction"):
|
||||||
|
with kanta.transaction(action="inc", log=custom_logger) as data:
|
||||||
|
data.counter = 1
|
||||||
|
|
||||||
|
await kanta.close()
|
||||||
|
|
||||||
|
info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
|
||||||
|
assert len(info_messages) >= 1
|
||||||
|
assert "inc" in info_messages[0].message
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_open_locked_file_raises_filelock_error(tmp_path, format_config):
|
async def test_open_locked_file_raises_filelock_error(tmp_path, format_config):
|
||||||
path = tmp_path / "test.db"
|
path = tmp_path / "test.db"
|
||||||
@@ -514,7 +838,7 @@ async def test_migrations_from_module_path(tmp_path, format_config):
|
|||||||
module_name = "test_migrations_path"
|
module_name = "test_migrations_path"
|
||||||
mod = type(sys)(module_name)
|
mod = type(sys)(module_name)
|
||||||
|
|
||||||
def migrate_v1(d, ctx):
|
def migrate_v1(d, kanta):
|
||||||
d["counter"] = 2
|
d["counter"] = 2
|
||||||
|
|
||||||
mod.__dict__["migrate_v1"] = migrate_v1
|
mod.__dict__["migrate_v1"] = migrate_v1
|
||||||
|
|||||||
+36
-5
@@ -1,16 +1,47 @@
|
|||||||
import logging
|
import logging
|
||||||
|
|
||||||
from kanta.logging import configure_logging, log_change
|
import pytest
|
||||||
from kanta.logging import logger
|
|
||||||
|
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()
|
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):
|
def test_log_change_no_diff(capsys):
|
||||||
logger.handlers.clear()
|
kanta_logger = logging.getLogger("kanta")
|
||||||
|
kanta_logger.handlers.clear()
|
||||||
configure_logging()
|
configure_logging()
|
||||||
log_change("test", {})
|
log_change("test", {})
|
||||||
captured = capsys.readouterr()
|
captured = capsys.readouterr()
|
||||||
|
|||||||
+171
-16
@@ -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():
|
def test_register_and_apply():
|
||||||
reg = MigrationRegistry()
|
reg = Migrations()
|
||||||
|
kanta = _DummyKanta()
|
||||||
|
|
||||||
@reg.register
|
@reg.register
|
||||||
def migrate_v1(d, ctx):
|
def migrate_v1(d, kanta):
|
||||||
d["version"] = 1
|
d["version"] = 1
|
||||||
|
|
||||||
@reg.register
|
@reg.register
|
||||||
def migrate_v2(d, ctx):
|
def migrate_v2(d, kanta):
|
||||||
d["version"] = 2
|
d["version"] = 2
|
||||||
|
|
||||||
state = {}
|
state = {}
|
||||||
new_ver = reg.apply(state, current_version=0, silent=True)
|
result = reg.apply(state, current_version=0, kanta=kanta)
|
||||||
assert new_ver == 2
|
assert result.version == 2
|
||||||
assert state["version"] == 2
|
assert state["version"] == 2
|
||||||
|
|
||||||
|
|
||||||
def test_no_migrations_needed():
|
def test_no_migrations_needed():
|
||||||
reg = MigrationRegistry()
|
reg = Migrations()
|
||||||
|
kanta = _DummyKanta()
|
||||||
|
|
||||||
@reg.register
|
@reg.register
|
||||||
def migrate_v1(d, ctx):
|
def migrate_v1(d, kanta):
|
||||||
d["x"] = 1
|
d["x"] = 1
|
||||||
|
|
||||||
state = {"x": 1}
|
state = {"x": 1}
|
||||||
new_ver = reg.apply(state, current_version=1, silent=True)
|
result = reg.apply(state, current_version=1, kanta=kanta)
|
||||||
assert new_ver == 1
|
assert result.version == 1
|
||||||
|
|
||||||
|
|
||||||
def test_from_module():
|
def test_from_module():
|
||||||
mod = ModuleType("fake_migrations")
|
mod = ModuleType("fake_migrations")
|
||||||
|
kanta = _DummyKanta()
|
||||||
|
|
||||||
def migrate_v1(d, ctx):
|
def migrate_v1(d, kanta):
|
||||||
d["v"] = 1
|
d["v"] = 1
|
||||||
|
|
||||||
def migrate_v2(d, ctx):
|
def migrate_v2(d, kanta):
|
||||||
d["v"] = 2
|
d["v"] = 2
|
||||||
|
|
||||||
mod.__dict__["migrate_v1"] = migrate_v1
|
mod.__dict__["migrate_v1"] = migrate_v1
|
||||||
mod.__dict__["migrate_v2"] = migrate_v2
|
mod.__dict__["migrate_v2"] = migrate_v2
|
||||||
|
|
||||||
reg = MigrationRegistry.from_module(mod)
|
reg = Migrations.from_module(mod)
|
||||||
assert reg.dbver == 2
|
assert reg.dbver == 2
|
||||||
|
|
||||||
state = {}
|
state = {}
|
||||||
new_ver = reg.apply(state, current_version=0, silent=True)
|
result = reg.apply(state, current_version=0, kanta=kanta)
|
||||||
assert new_ver == 2
|
assert result.version == 2
|
||||||
assert state["v"] == 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
@@ -82,8 +82,9 @@ async def test_transaction_mtime_false_preserves_mtime(tmp_path, format_config):
|
|||||||
continue
|
continue
|
||||||
records.append(serializer.decode(payload, type=ChangeRecord))
|
records.append(serializer.decode(payload, type=ChangeRecord))
|
||||||
|
|
||||||
assert records[0].m == first_m
|
assert records[0].a == "bootstrap"
|
||||||
assert records[1].m is None
|
assert records[1].m == first_m
|
||||||
|
assert records[2].m is None
|
||||||
assert kanta.mtime == first_m
|
assert kanta.mtime == first_m
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -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()
|
||||||
@@ -32,3 +32,20 @@ def test_force_writes():
|
|||||||
f = FakeFile()
|
f = FakeFile()
|
||||||
ss.maybe_write(f, 1, {"x": 1})
|
ss.maybe_write(f, 1, {"x": 1})
|
||||||
assert len(f.written) == 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