LeoVasanko/kanta
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement proper bootstrap and fatal error handling via decorators. Breaks compatibility.

Browse Source
This commit is contained in:
LeoVasanko
2026-06-13 02:53:09 +00:00
parent acabc049e7
commit 4db046d627
6 changed files with 408 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+58
View File
@@ -51,6 +51,64 @@ asyncio.run(main())
3. Let Kanta flush queued changes to disk in the background.
4. Use snapshots and replay for fast startup and full history.
## Bootstrap and Open Modes
Kanta supports open-time bootstrap callbacks for initializing a brand-new
database before `open()` returns.
Register bootstrap handlers with a decorator:
```python
kanta = Kanta("data.kantadb", Data())
@kanta.bootstrap(action="seed", user="system")
def seed_defaults(data) -> None:
data.users["admin"] = User(name="Admin")
await kanta.open()
```
You can also use `@kanta.bootstrap` with no arguments and async handlers:
```python
@kanta.bootstrap
async def bootstrap_async(data) -> None:
data.counter = 1
```
When multiple bootstrap handlers are registered:
- they run in registration order,
- exactly one bootstrap change record is queued,
- bootstrap metadata (`action`, `user`, `mtime`) is taken from the last
registration.
If any bootstrap handler raises, Kanta closes and removes the database file,
then re-raises the error.
`open()` also supports strict open mode:
```python
await kanta.open(create=False)
```
With `create=False`, open fails if the database file does not exist or is
empty.
## Fatal Error Handlers
Fatal background write errors can be observed with a decorator:
```python
import os
import signal
@kanta.fatal_error
async def on_fatal(err):
os.kill(os.getpid(), signal.SIGTERM) # Die
```
Multiple fatal handlers are supported and run in registration order.
## Migrations
Adding or removing a field and other such simple operations are automatic, but when the time comes to really change your data model, implement a `migrate_v1` function that converts your old data to the new form. This works on plain built-in dict and other types, to avoid needing to preserve old versions of your structs.
docs/database.md
+27
View File
@@ -114,6 +114,33 @@ reloads, while system operations such as migrations leave it unchanged.
- `kanta.close()` performs final flush and releases file resources.
- `async with Kanta(...)` guarantees open/close lifecycle management.
### Open Modes
- `await kanta.open()` (default) creates the database file if missing.
- `await kanta.open(create=False)` fails when the file is missing or empty.
### Bootstrap Callbacks
- Bootstrap callbacks run during `open()` when the database is empty.
- Register callbacks via:
- `@kanta.bootstrap`
- `@kanta.bootstrap(action=..., user=..., mtime=...)`
- Bootstrap callbacks may be sync or async and receive the live root data
object.
- Multiple bootstrap callbacks are supported:
- callbacks execute in registration order,
- exactly one bootstrap `ChangeRecord` is queued,
- bootstrap metadata (`action`, `user`, `mtime`) is taken from the last
callback registration.
- If any bootstrap callback raises, Kanta closes and removes the database file,
then re-raises the exception.
### Fatal Error Handlers
- Fatal background persistence errors can be handled with `@kanta.fatal_error`.
- Handlers may be sync or async.
- Multiple handlers are supported and invoked in registration order.
## Migrations
- Migration source is configured on `Kanta(...)` via `migrations=`.
kanta/kanta.py
+53 -8
View File
@@ -1,8 +1,6 @@
"""JSONL persistence layer with background flush task."""
from __future__ import annotations
from collections.abc import Callable
from datetime import datetime
from pathlib import Path
from types import ModuleType
@@ -55,7 +53,6 @@ class Kanta(Generic[T]):
migrations: ModuleType | str | None = None,
migration_ctx: Any | None = None,
serializer: Serializer | None = None,
fatal_error: Callable[[DatabaseError], None] | None = None,
flush_interval: float = 0.1,
):
"""Initialize a Kanta persistence instance.
@@ -68,8 +65,6 @@ class Kanta(Generic[T]):
migration_ctx: Optional context object passed to migration functions.
flush_interval: Background flush interval in seconds.
serializer: Optional serializer implementation.
fatal_error: Optional callback invoked immediately when the
background writer encounters a DatabaseError.
Raises:
ImportError: If ``migrations`` is a string path that cannot be imported.
@@ -80,7 +75,6 @@ class Kanta(Generic[T]):
self._impl = KantaImpl(
serializer=active_serializer,
fatal_error=fatal_error,
filename=filename,
data=data,
type=data_type,
@@ -144,19 +138,23 @@ class Kanta(Generic[T]):
"""
return self._impl.mtime
async def open(self) -> None:
async def open(self, *, create: bool = True) -> None:
"""Open the database file and start background persistence.
This loads existing records, applies configured migrations, and starts
the background flush task.
Args:
create: Whether to create the database file when missing.
If False, opening fails when the file does not exist or is empty.
Calling ``open`` more than once on the same instance is not allowed.
Raises:
kanta.exceptions.DatabaseError: If replay or decoding fails.
kanta.exceptions.DataIntegrityError: If the instance is already open.
"""
await self._impl.open()
await self._impl.open(create=create)
async def __aenter__(self) -> Kanta[T]:
"""Enter async context manager and open the database.
@@ -189,6 +187,53 @@ class Kanta(Generic[T]):
"""Stop background task, flush pending changes, and close file lock."""
await self._impl.close()
def bootstrap(
self,
fn=None,
*,
action: str = "bootstrap",
user: str | None = None,
mtime: bool | datetime = True,
):
"""Register a bootstrap callback executed during :meth:`open`.
Can be used as ``@kanta.bootstrap`` or ``@kanta.bootstrap(...)``.
The callback receives the live ``data`` object and may be sync or async.
"""
def _register(callback):
if not callable(callback):
raise TypeError("bootstrap callback must be callable")
self._impl.add_bootstrap(
callback=callback,
action=action,
user=user,
mtime=mtime,
)
return callback
if fn is None:
return _register
return _register(fn)
def fatal_error(self, fn=None):
"""Register fatal error handler callback.
Can be used as ``@kanta.fatal_error``.
The callback receives a :class:`kanta.exceptions.DatabaseError` and may
be sync or async.
"""
def _register(callback):
if not callable(callback):
raise TypeError("fatal error callback must be callable")
self._impl.add_fatal_error(callback)
return callback
if fn is None:
return _register
return _register(fn)
def transaction(
self,
action: str,
kanta/kantaimpl.py
+57 -2
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import asyncio
import copy
import importlib
import inspect
import logging
from datetime import UTC, datetime
from typing import Any, Generic, TypeVar
@@ -41,13 +42,31 @@ class KantaImpl(PersistenceMixin, Generic[T]):
self.in_transaction = False
self.transaction_snapshot: dict[str, Any] | None = None
self.opened = False
self.bootstrap_callbacks: list[Any] = []
self.bootstrap_action = "bootstrap"
self.bootstrap_user: str | None = None
self.bootstrap_mtime: bool | datetime = True
self.statedict = struct_to_dict(self.data, serializer=self.serializer)
self.version = (
self.migration_registry.dbver if self.migration_registry is not None else 0
)
async def open(self) -> None:
def add_bootstrap(
self,
*,
callback,
action: str,
user: str | None,
mtime: bool | datetime,
) -> None:
"""Add bootstrap callback and update bootstrap metadata."""
self.bootstrap_callbacks.append(callback)
self.bootstrap_action = action
self.bootstrap_user = user
self.bootstrap_mtime = mtime
async def open(self, *, create: bool = True) -> None:
"""Open the database: load from disk, apply migrations, start background task."""
if self.opened:
raise DataIntegrityError(
@@ -56,10 +75,25 @@ class KantaImpl(PersistenceMixin, Generic[T]):
action="open",
)
existed_before_open = self.filename.exists()
content = await asyncio.to_thread(
self.file.open_and_read,
self.filename,
create=True,
create=create,
)
if not create and (not existed_before_open or not content):
self.file.close()
reason = (
"database file did not exist"
if not existed_before_open
else "database file is empty"
)
raise DataIntegrityError(
f"Cannot open database: {reason}",
db_path=self.filename,
action="open",
)
if content:
@@ -112,6 +146,27 @@ class KantaImpl(PersistenceMixin, Generic[T]):
if rr.last_snapshot_mtime is not None
else None
)
elif self.bootstrap_callbacks:
try:
for callback in self.bootstrap_callbacks:
callback_result = callback(self.data)
if inspect.isawaitable(callback_result):
await callback_result
current = struct_to_dict(self.data, serializer=self.serializer)
self.queue_change(
self.bootstrap_action,
current,
user=self.bootstrap_user,
mtime=self.bootstrap_mtime,
)
except Exception:
self.file.close()
try:
await asyncio.to_thread(self.filename.unlink, missing_ok=True)
except FileNotFoundError:
pass
raise
self.opened = True
kanta/persistence.py
+11 -5
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import asyncio
import copy
import inspect
import logging
from collections import deque
from collections.abc import Callable
@@ -34,7 +35,7 @@ class PersistenceMixin:
serializer: Serializer
framer: Framer
background_task: asyncio.Task | None
fatal_error: Callable[[DatabaseError], None] | None
fatal_error_handlers: list[Callable[[DatabaseError], Any]]
background_error: DatabaseError | None
flush_interval: float
version: int
@@ -46,7 +47,6 @@ class PersistenceMixin:
filename = kwargs.pop("filename")
flush_interval = kwargs.pop("flush_interval", 0.1)
serializer = kwargs.pop("serializer", None)
fatal_error = kwargs.pop("fatal_error", None)
super().__init__(**kwargs)
self.filename = Path(filename)
self.file = LockedFile()
@@ -57,12 +57,16 @@ class PersistenceMixin:
self.framer = self.serializer.framer_cls()
self.snapshot = SnapshotState(serializer=self.serializer, framer=self.framer)
self.background_task = None
self.fatal_error = fatal_error
self.fatal_error_handlers = []
self.background_error = None
self.flush_interval = flush_interval
self.version = 0
self.mtime: datetime | None = None
def add_fatal_error(self, callback: Callable[[DatabaseError], Any]) -> None:
"""Register one fatal error callback in call order."""
self.fatal_error_handlers.append(callback)
async def _background_loop(self) -> None:
"""Background task that periodically flushes changes to disk."""
while True:
@@ -76,9 +80,11 @@ class PersistenceMixin:
break
except DatabaseError as e:
self.background_error = e
if self.fatal_error is not None:
for callback in self.fatal_error_handlers:
try:
self.fatal_error(e)
callback_result = callback(e)
if inspect.isawaitable(callback_result):
await callback_result
except Exception as callback_error:
_logger.exception(
"Background error callback failed: %s", callback_error
tests/test_kanta_integration.py
+202 -5
View File
@@ -100,6 +100,202 @@ async def test_bootstrap_creates_file(tmp_path, format_config):
assert path.exists()
@pytest.mark.asyncio
async def test_bootstrap_decorator_with_args(tmp_path, format_config):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
@kanta.bootstrap(action="seed_init", user="system")
def seed(data):
data.counter = 3
await kanta.open()
await kanta.close()
assert change_actions(path, format_config) == ["seed_init"]
@pytest.mark.asyncio
async def test_bootstrap_decorator_without_args(tmp_path, format_config):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
@kanta.bootstrap
def seed(data):
data.counter = 4
await kanta.open()
await kanta.close()
assert change_actions(path, format_config) == ["bootstrap"]
@pytest.mark.asyncio
async def test_bootstrap_decorator_async(tmp_path, format_config):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
@kanta.bootstrap(action="async_seed")
async def seed(data):
await asyncio.sleep(0)
data.counter = 5
await kanta.open()
await kanta.close()
assert change_actions(path, format_config) == ["async_seed"]
@pytest.mark.asyncio
async def test_bootstrap_decorator_multiple_handlers_in_order(tmp_path, format_config):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
@kanta.bootstrap(action="boot_1")
def seed_one(data):
data.counter = 1
@kanta.bootstrap(action="boot_2")
async def seed_two(data):
await asyncio.sleep(0)
data.counter = 2
await kanta.open()
await kanta.close()
assert change_actions(path, format_config) == ["boot_2"]
@pytest.mark.asyncio
async def test_bootstrap_failure_removes_database_file(tmp_path, format_config):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
@kanta.bootstrap(action="boot_fail")
def seed_fail(data):
data.counter = 10
raise RuntimeError("bootstrap failed")
with pytest.raises(RuntimeError, match="bootstrap failed"):
await kanta.open()
assert not path.exists()
@pytest.mark.asyncio
async def test_bootstrap_async_failure_removes_database_file(tmp_path, format_config):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
@kanta.bootstrap(action="boot_fail_async")
async def seed_fail(data):
await asyncio.sleep(0)
data.counter = 10
raise RuntimeError("bootstrap async failed")
with pytest.raises(RuntimeError, match="bootstrap async failed"):
await kanta.open()
assert not path.exists()
@pytest.mark.asyncio
async def test_open_create_false_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(create=False)
@pytest.mark.asyncio
async def test_open_create_false_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(create=False)
@pytest.mark.asyncio
async def test_background_write_failure_notifies_decorator_callback(
tmp_path, format_config, monkeypatch
):
path = tmp_path / "test.db"
errors: list[DatabaseError] = []
signaled = asyncio.Event()
kanta = make_kanta(
path,
Data,
format_config,
flush_interval=0.01,
)
@kanta.fatal_error
async def on_fatal_error(err: DatabaseError) -> None:
errors.append(err)
signaled.set()
await kanta.open()
with kanta.transaction(action="inc") as data:
data.counter = 1
def fail_write(_data: bytes) -> None:
raise OSError("simulated background write failure")
monkeypatch.setattr(kanta._impl.file, "write", fail_write)
await asyncio.wait_for(signaled.wait(), timeout=1.0)
assert errors
assert "Failed to flush database" in str(errors[0])
await kanta.close()
@pytest.mark.asyncio
async def test_background_write_failure_notifies_multiple_callbacks_in_order(
tmp_path, format_config, monkeypatch
):
path = tmp_path / "test.db"
calls: list[str] = []
signaled = asyncio.Event()
kanta = make_kanta(
path,
Data,
format_config,
flush_interval=0.01,
)
@kanta.fatal_error
def on_fatal_error_sync(err: DatabaseError) -> None:
calls.append("sync")
@kanta.fatal_error
async def on_fatal_error_async(err: DatabaseError) -> None:
await asyncio.sleep(0)
calls.append("async")
signaled.set()
await kanta.open()
with kanta.transaction(action="inc") as data:
data.counter = 1
def fail_write(_data: bytes) -> None:
raise OSError("simulated background write failure")
monkeypatch.setattr(kanta._impl.file, "write", fail_write)
await asyncio.wait_for(signaled.wait(), timeout=1.0)
assert calls == ["sync", "async"]
await kanta.close()
@pytest.mark.asyncio
async def test_snapshot(tmp_path, format_config):
path = tmp_path / "test.db"
@@ -281,17 +477,18 @@ async def test_background_write_failure_notifies_callback(
errors: list[DatabaseError] = []
signaled = asyncio.Event()
def on_fatal_error(err: DatabaseError) -> None:
errors.append(err)
signaled.set()
kanta = make_kanta(
path,
Data,
format_config,
flush_interval=0.01,
fatal_error=on_fatal_error,
)
@kanta.fatal_error
def on_fatal_error(err: DatabaseError) -> None:
errors.append(err)
signaled.set()
await kanta.open()
with kanta.transaction(action="inc") as data: