From 4db046d627a440d43d0214202507d77a234f8565 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Sat, 13 Jun 2026 02:53:09 +0000 Subject: [PATCH] Implement proper bootstrap and fatal error handling via decorators. Breaks compatibility. --- README.md | 58 +++++++++ docs/database.md | 27 +++++ kanta/kanta.py | 61 ++++++++-- kanta/kantaimpl.py | 59 ++++++++- kanta/persistence.py | 16 ++- tests/test_kanta_integration.py | 207 +++++++++++++++++++++++++++++++- 6 files changed, 408 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index af10369..86db3e1 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/docs/database.md b/docs/database.md index e920603..394b3ef 100644 --- a/docs/database.md +++ b/docs/database.md @@ -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=`. diff --git a/kanta/kanta.py b/kanta/kanta.py index 17ff4d4..a029d5b 100644 --- a/kanta/kanta.py +++ b/kanta/kanta.py @@ -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, diff --git a/kanta/kantaimpl.py b/kanta/kantaimpl.py index e97fa74..c298fac 100644 --- a/kanta/kantaimpl.py +++ b/kanta/kantaimpl.py @@ -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,12 +75,27 @@ 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: try: rr = replay( @@ -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 diff --git a/kanta/persistence.py b/kanta/persistence.py index 6604d0e..64f1214 100644 --- a/kanta/persistence.py +++ b/kanta/persistence.py @@ -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 diff --git a/tests/test_kanta_integration.py b/tests/test_kanta_integration.py index 18e9084..19ff268 100644 --- a/tests/test_kanta_integration.py +++ b/tests/test_kanta_integration.py @@ -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: