2 Commits
13 changed files with 413 additions and 96 deletions
+12
View File
@@ -94,6 +94,18 @@ await kanta.open(create=False)
With `create=False`, open fails if the database file does not exist or is
empty.
Read-only mode opens an existing database without locking it or starting the
background flush task. This is useful for readers that must not block the
writer or modify the file:
```python
await kanta.open(readonly=True)
```
In read-only mode, records are replayed and migrations are applied in memory,
but transactions and explicit flushes are rejected and the file is never
created if missing.
## Fatal Error Handlers
Fatal background write errors can be observed with a decorator:
+6
View File
@@ -118,6 +118,12 @@ reloads, while system operations such as migrations leave it unchanged.
- `await kanta.open()` (default) creates the database file if missing.
- `await kanta.open(create=False)` fails when the file is missing or empty.
- `await kanta.open(readonly=True)` opens an existing database read-only.
- The file is opened without acquiring a lock and without a background flush
task.
- Existing records are replayed and migrations are still applied in memory.
- Transactions and explicit flushes are rejected.
- The file is never created if missing.
### Callbacks
+18 -25
View File
@@ -12,6 +12,7 @@ and receive the value plus an optional ``path`` string. They return
from __future__ import annotations
import inspect
import types
from collections.abc import Callable
from dataclasses import dataclass
from typing import Annotated, Any, Union, get_args, get_origin
@@ -317,17 +318,13 @@ class CallbackRegistry:
f"Allowed: str path, {self._allowed_message('logfmt')}"
)
if sig.return_annotation is inspect.Signature.empty:
raise TypeError(
f"logfmt callback {callback.__name__} must annotate its "
f"return type as str | None"
)
return_ann = self._resolve_raw_annotation(sig.return_annotation, callback)
if not self._is_optional_str(return_ann):
raise TypeError(
f"logfmt callback {callback.__name__} must return str | None, "
f"got {return_ann!r}"
)
if sig.return_annotation is not inspect.Signature.empty:
return_ann = self._resolve_raw_annotation(sig.return_annotation, callback)
if not self._is_optional_str(return_ann):
raise TypeError(
f"logfmt callback {callback.__name__} must return str | None, "
f"got {return_ann!r}"
)
return _LogFmtFunctionSpec(
callback=callback,
@@ -416,19 +413,15 @@ class CallbackRegistry:
f"logfmt class {cls.__name__}.resolve must accept a 'path: str' parameter"
)
if resolve_sig.return_annotation is inspect.Signature.empty:
raise TypeError(
f"logfmt class {cls.__name__}.resolve must annotate its "
f"return type as str | None"
)
return_ann = self._resolve_raw_annotation(
resolve_sig.return_annotation, resolve
)
if not self._is_optional_str(return_ann):
raise TypeError(
f"logfmt class {cls.__name__}.resolve must return str | None, "
f"got {return_ann!r}"
if resolve_sig.return_annotation is not inspect.Signature.empty:
return_ann = self._resolve_raw_annotation(
resolve_sig.return_annotation, resolve
)
if not self._is_optional_str(return_ann):
raise TypeError(
f"logfmt class {cls.__name__}.resolve must return str | None, "
f"got {return_ann!r}"
)
return _LogFmtClassSpec(cls=cls, inject_params=inject_params, path=path)
@@ -516,7 +509,7 @@ class CallbackRegistry:
@staticmethod
def _unwrap_optional(ann: Any) -> Any:
origin = get_origin(ann)
if origin is not Union:
if origin not in (Union, types.UnionType):
return ann
args = [arg for arg in get_args(ann) if arg is not type(None)]
return args[0] if len(args) == 1 else ann
@@ -524,7 +517,7 @@ class CallbackRegistry:
@staticmethod
def _is_optional_str(ann: Any) -> bool:
origin = get_origin(ann)
if origin is not Union:
if origin not in (Union, types.UnionType):
return ann is str
args = get_args(ann)
return type(None) in args and any(arg is str for arg in args)
+40 -26
View File
@@ -34,6 +34,7 @@ if sys.platform == "win32":
_GENERIC_READ = 0x80000000
_GENERIC_WRITE = 0x40000000
_FILE_SHARE_READ = 0x00000001
_FILE_SHARE_WRITE = 0x00000002
_OPEN_EXISTING = 3
_OPEN_ALWAYS = 4
_FILE_ATTRIBUTE_NORMAL = 0x80
@@ -91,15 +92,16 @@ else:
class LockedFile:
"""A file opened with an exclusive write lock.
"""A file opened for read+write with an optional exclusive lock.
Usage::
f = LockedFile()
f.open(path) # open + lock (read+write)
content = f.read() # read entire content
f.write(data) # append data (seeks to end first)
f.close() # release lock + close fd
f.open(path) # open + lock (read+write)
f.open(path, readonly=True) # open read-only without locking
content = f.read() # read entire content
f.write(data) # append data (seeks to end first)
f.close() # release lock + close fd
Unix: fcntl.flock (advisory) — read-only callers that don't flock are unaffected.
Windows: CreateFileW with FILE_SHARE_READ — OS blocks other writers.
@@ -108,12 +110,13 @@ class LockedFile:
def __init__(self) -> None:
self._fd: int | None = None # Unix fd or Windows HANDLE
def open(self, path: Path, *, create: bool = False) -> None:
"""Open *path* for read+write with an exclusive lock.
def open(self, path: Path, *, create: bool = False, readonly: bool = False) -> None:
"""Open *path* and optionally acquire an exclusive lock.
Args:
path: File to open and lock.
create: If True, create the file if it doesn't exist (bootstrap).
readonly: If True, open read-only without acquiring a lock.
Raises:
FileLockError: If the file is locked by another process or not found.
@@ -122,16 +125,16 @@ class LockedFile:
return # Already open (idempotent)
if sys.platform == "win32":
self._open_win32(path, create)
self._open_win32(path, create, readonly)
else:
self._open_unix(path, create)
self._open_unix(path, create, readonly)
def open_and_read(self, path: Path, create: bool = False) -> bytes:
"""Open *path* with exclusive lock and read all content.
def open_and_read(self, path: Path, create: bool = False, readonly: bool = False) -> bytes:
"""Open *path* and read all content.
Combined operation for efficient use with asyncio.to_thread().
"""
self.open(path, create=create)
self.open(path, create=create, readonly=readonly)
return self.read()
def read(self) -> bytes:
@@ -188,20 +191,24 @@ class LockedFile:
# -- Unix ----------------------------------------------------------------
def _open_unix(self, path: Path, create: bool) -> None:
flags = os.O_RDWR | (os.O_CREAT if create else 0)
def _open_unix(self, path: Path, create: bool, readonly: bool) -> None:
if readonly:
flags = os.O_RDONLY
else:
flags = os.O_RDWR | (os.O_CREAT if create else 0)
try:
fd = os.open(path, flags, 0o666)
except FileNotFoundError:
_fatal(f"Database file not found: {path.resolve()}", db_path=path)
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
os.close(fd)
_fatal(
f"{path.resolve()}: database already locked by another instance",
db_path=path,
)
if not readonly:
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
os.close(fd)
_fatal(
f"{path.resolve()}: database already locked by another instance",
db_path=path,
)
self._fd = fd
def _read_unix(self) -> bytes:
@@ -220,12 +227,19 @@ class LockedFile:
# -- Windows -------------------------------------------------------------
def _open_win32(self, path: Path, create: bool) -> None:
disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING
def _open_win32(self, path: Path, create: bool, readonly: bool) -> None:
if readonly:
disposition = _OPEN_EXISTING
access = _GENERIC_READ
share = _FILE_SHARE_READ | _FILE_SHARE_WRITE
else:
disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING
access = _GENERIC_READ | _GENERIC_WRITE
share = _FILE_SHARE_READ
handle = _kernel32.CreateFileW(
str(path),
_GENERIC_READ | _GENERIC_WRITE,
_FILE_SHARE_READ,
access,
share,
None,
disposition,
_FILE_ATTRIBUTE_NORMAL,
+15 -6
View File
@@ -3,7 +3,7 @@
from __future__ import annotations
from datetime import datetime
from pathlib import Path
from types import ModuleType
from types import ModuleType, SimpleNamespace
from typing import Any, Generic, TypeVar
from kanta.kantaimpl import KantaImpl
@@ -50,7 +50,6 @@ class Kanta(Generic[T]):
*,
type: type[T] | None = None,
migrations: ModuleType | str | None = None,
migration_ctx: Any | None = None,
serializer: Serializer | None = None,
flush_interval: float = 0.1,
):
@@ -61,7 +60,6 @@ class Kanta(Generic[T]):
data: Caller-owned root msgspec.Struct state instance.
type: Optional explicit root type. Defaults to ``type(data)``.
migrations: Optional migrations module object or import path.
migration_ctx: Optional context object passed to migration functions.
flush_interval: Background flush interval in seconds.
serializer: Optional serializer implementation.
@@ -78,7 +76,6 @@ class Kanta(Generic[T]):
data=data,
type=data_type,
migrations=migrations,
migration_ctx=migration_ctx,
flush_interval=flush_interval,
kanta=self,
)
@@ -127,6 +124,15 @@ class Kanta(Generic[T]):
"""
return self._impl.filename
@property
def ctx(self) -> SimpleNamespace:
"""User-writable context namespace.
Migration functions receive the ``Kanta`` instance and can read or
mutate ``kanta.ctx`` during migrations.
"""
return self._impl.ctx
@property
def mtime(self) -> datetime | None:
"""Last modification time carried forward from change records.
@@ -138,7 +144,7 @@ class Kanta(Generic[T]):
"""
return self._impl.mtime
async def open(self, *, create: bool = True) -> None:
async def open(self, *, create: bool = True, readonly: bool = False) -> None:
"""Open the database file and start background persistence.
This loads existing records, applies configured migrations, and starts
@@ -147,6 +153,9 @@ class Kanta(Generic[T]):
Args:
create: Whether to create the database file when missing.
If False, opening fails when the file does not exist or is empty.
readonly: If True, open the database read-only. No lock is acquired,
no background flush task is started, and transactions are
rejected. The file is not created if missing.
Calling ``open`` more than once on the same instance is not allowed.
@@ -154,7 +163,7 @@ class Kanta(Generic[T]):
kanta.exceptions.DatabaseError: If replay or decoding fails.
kanta.exceptions.DataIntegrityError: If the instance is already open.
"""
await self._impl.open(create=create)
await self._impl.open(create=create, readonly=readonly)
async def __aenter__(self) -> Kanta[T]:
"""Enter async context manager and open the database.
+26 -7
View File
@@ -7,6 +7,7 @@ import copy
import importlib
import logging
from datetime import UTC, datetime
from types import SimpleNamespace
from typing import Any, Generic, TypeVar
from kanta.callbacks import CallbackRegistry, InjectionContext
@@ -28,8 +29,8 @@ class KantaImpl(PersistenceMixin, Generic[T]):
self.data_type = kwargs.pop("type")
self.data: T = kwargs.pop("data")
self.migrations = kwargs.pop("migrations", None)
self.migration_ctx = kwargs.pop("migration_ctx", None)
self._kanta = kwargs.pop("kanta", None)
self.ctx = SimpleNamespace()
super().__init__(**kwargs)
self.migration_registry: MigrationRegistry | None = None
if self.migrations is not None:
@@ -43,6 +44,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
self.in_transaction = False
self.transaction_snapshot: dict[str, Any] | None = None
self.opened = False
self.readonly = False
self.bootstrap_action = "bootstrap"
self.bootstrap_user: str | None = None
self.bootstrap_mtime: bool | datetime = True
@@ -75,7 +77,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
"""Register one transaction logfmt callback."""
self.callback_registry.register("logfmt", callback, path=path)
async def open(self, *, create: bool = True) -> None:
async def open(self, *, create: bool = True, readonly: bool = False) -> None:
"""Open the database: load from disk, apply migrations, start background task."""
if self.opened:
raise DataIntegrityError(
@@ -84,12 +86,17 @@ class KantaImpl(PersistenceMixin, Generic[T]):
action="open",
)
self.readonly = readonly
existed_before_open = self.filename.exists()
# Read-only mode never creates the file.
open_create = create and not readonly
content = await asyncio.to_thread(
self.file.open_and_read,
self.filename,
create=create,
create=open_create,
readonly=readonly,
)
if not create and (not existed_before_open or not content):
@@ -136,7 +143,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
if self.migration_registry is not None:
rr.version = self.migration_registry.apply(
rr.state, rr.version, self.migration_ctx
rr.state, rr.version, self._kanta
)
self.statedict = copy.deepcopy(rr.state)
@@ -149,12 +156,22 @@ class KantaImpl(PersistenceMixin, Generic[T]):
self.version = rr.version
self.mtime = rr.m
normalized = struct_to_dict(self.data, serializer=self.serializer)
self.queue_change("migrate:msgspec", normalized, mtime=False)
if self.readonly:
self.statedict = copy.deepcopy(normalized)
else:
self.queue_change("migrate:msgspec", normalized, mtime=False)
self.snapshot.ts = (
datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)
if rr.last_snapshot_mtime is not None
else None
)
elif self.readonly:
self.file.close()
raise DataIntegrityError(
"Cannot open empty database in read-only mode",
db_path=self.filename,
action="open",
)
elif self.callback_registry.has("bootstrap"):
try:
await self.callback_registry.invoke(
@@ -179,7 +196,8 @@ class KantaImpl(PersistenceMixin, Generic[T]):
self.opened = True
self.background_task = asyncio.create_task(self._background_loop())
if not self.readonly:
self.background_task = asyncio.create_task(self._background_loop())
async def close(self) -> None:
"""Stop the background task, flush pending changes, and release the file lock."""
@@ -196,7 +214,8 @@ class KantaImpl(PersistenceMixin, Generic[T]):
# Always run a final flush in case the background task never reached
# its cancellation handler.
await self.flush()
if not self.readonly:
await self.flush()
self.file.close()
self.opened = False
+21 -16
View File
@@ -7,24 +7,14 @@ or by prefix. Each runs exactly once based on the current version.
from __future__ import annotations
import importlib
import inspect
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.
@@ -33,15 +23,20 @@ class MigrationRegistry:
registry = MigrationRegistry()
@registry.register
def migrate_v1(d: dict, ctx: MigrationCtx) -> None:
def migrate_v1(d: dict, kanta) -> None:
d.setdefault("version", 1)
kanta.ctx.note = "migrated"
new_version = registry.apply(state, current_version=0)
@registry.register
def migrate_v2(d: dict) -> None:
d.setdefault("version", 2)
new_version = registry.apply(state, current_version=0, kanta=kanta)
Or load from a module::
registry = MigrationRegistry.from_module("myapp.migrations")
new_version = registry.apply(state, current_version=0)
new_version = registry.apply(state, current_version=0, kanta=kanta)
"""
def __init__(self) -> None:
@@ -89,11 +84,21 @@ class MigrationRegistry:
"""Current schema version (= highest discovered migration, or 0)."""
return max(self._migrations.keys(), default=0)
@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,
ctx: MigrationCtx | None = None,
kanta: Any,
*,
silent: bool = False,
) -> int:
@@ -109,7 +114,7 @@ class MigrationRegistry:
f"Missing migration step migrate_v{next_version} "
f"(highest discovered is v{self.dbver})"
)
fn(data_dict, ctx or MigrationCtx())
self._call_migration(fn, data_dict, kanta)
current_version = next_version
if not silent:
desc = (fn.__doc__ or fn.__name__).split("\n")[0].rstrip(".")
+17
View File
@@ -39,6 +39,7 @@ class PersistenceMixin:
flush_interval: float
version: int
opened: bool
readonly: bool
mtime: datetime | None
def __init__(self, **kwargs: Any) -> None:
@@ -68,6 +69,8 @@ class PersistenceMixin:
async def _background_loop(self) -> None:
"""Background task that periodically flushes changes to disk."""
if self.readonly:
return
while True:
try:
await asyncio.sleep(self.flush_interval)
@@ -160,6 +163,13 @@ class PersistenceMixin:
action="flush_sync",
)
if self.readonly:
raise DataIntegrityError(
"Cannot flush in read-only mode",
db_path=self.filename,
action="flush_sync",
)
if self.flush_failed:
return
@@ -207,6 +217,13 @@ class PersistenceMixin:
action="flush",
)
if self.readonly:
raise DataIntegrityError(
"Cannot flush in read-only mode",
db_path=self.filename,
action="flush",
)
if self.flush_failed:
return
+7
View File
@@ -24,6 +24,13 @@ def transaction(
mtime: bool | datetime = True,
):
"""Wrap writes in a transaction and yield the live db object."""
if impl.readonly:
raise DataIntegrityError(
"Cannot start transaction in read-only mode",
db_path=impl.filename,
action=action,
)
if impl.in_transaction:
raise RuntimeError(
"Nested or simultaneous transactions are not supported "
+49 -5
View File
@@ -1,4 +1,4 @@
from typing import Any
from typing import Any, Optional, Union
import pytest
@@ -49,16 +49,60 @@ def test_logfmt_requires_value_annotation(tmp_path, format_config):
return None
def test_logfmt_requires_return_annotation(tmp_path, format_config):
def test_logfmt_allows_missing_return_annotation(tmp_path, format_config):
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
with pytest.raises(TypeError, match="must annotate its return"):
@kanta.logfmt
def resolve_names(value: str, current: DictPost):
return None
@kanta.logfmt
def resolve_names(value: str, current: DictPost):
def test_logfmt_class_allows_missing_return_annotation(tmp_path, format_config):
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
@kanta.logfmt
class UserLogFmt(LogFmt):
def resolve(self, value: str, path: str):
return None
# fmt: off
def test_logfmt_accepts_optional_return_typing_forms(tmp_path, format_config):
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
@kanta.logfmt
def resolve_optional(value: str) -> Optional[str]: # noqa: UP007
return value
@kanta.logfmt
def resolve_union(value: str) -> Union[str, None]: # noqa: UP007
return value
@kanta.logfmt
def resolve_pipe(value: "str") -> "str | None":
return value
def test_logfmt_class_accepts_optional_return_typing_forms(tmp_path, format_config):
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
@kanta.logfmt
class OptionalStyle(LogFmt):
def resolve(self, value: str, path: str) -> Optional[str]: # noqa: UP007
return value
@kanta.logfmt
class UnionStyle(LogFmt):
def resolve(self, value: str, path: str) -> Union[str, None]: # noqa: UP007
return value
@kanta.logfmt
class StringStyle(LogFmt):
def resolve(self, value: "str", path: "str") -> "str | None":
return value
# fmt: on
def test_logfmt_rejects_async_callback(tmp_path, format_config):
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+2 -2
View File
@@ -404,7 +404,7 @@ async def test_migrations_from_module(tmp_path, format_config):
mod = type(sys)("test_migrations")
def migrate_v1(d, ctx):
def migrate_v1(d, kanta):
d["version"] = 1
mod.__dict__["migrate_v1"] = migrate_v1
@@ -514,7 +514,7 @@ async def test_migrations_from_module_path(tmp_path, format_config):
module_name = "test_migrations_path"
mod = type(sys)(module_name)
def migrate_v1(d, ctx):
def migrate_v1(d, kanta):
d["counter"] = 2
mod.__dict__["migrate_v1"] = migrate_v1
+47 -9
View File
@@ -1,44 +1,52 @@
from types import ModuleType
from types import ModuleType, SimpleNamespace
from kanta.migrate import MigrationRegistry
class _DummyKanta:
def __init__(self):
self.ctx = SimpleNamespace()
def test_register_and_apply():
reg = MigrationRegistry()
kanta = _DummyKanta()
@reg.register
def migrate_v1(d, ctx):
def migrate_v1(d, kanta):
d["version"] = 1
@reg.register
def migrate_v2(d, ctx):
def migrate_v2(d, kanta):
d["version"] = 2
state = {}
new_ver = reg.apply(state, current_version=0, silent=True)
new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
assert new_ver == 2
assert state["version"] == 2
def test_no_migrations_needed():
reg = MigrationRegistry()
kanta = _DummyKanta()
@reg.register
def migrate_v1(d, ctx):
def migrate_v1(d, kanta):
d["x"] = 1
state = {"x": 1}
new_ver = reg.apply(state, current_version=1, silent=True)
new_ver = reg.apply(state, current_version=1, kanta=kanta, silent=True)
assert new_ver == 1
def test_from_module():
mod = ModuleType("fake_migrations")
kanta = _DummyKanta()
def migrate_v1(d, ctx):
def migrate_v1(d, kanta):
d["v"] = 1
def migrate_v2(d, ctx):
def migrate_v2(d, kanta):
d["v"] = 2
mod.__dict__["migrate_v1"] = migrate_v1
@@ -48,6 +56,36 @@ def test_from_module():
assert reg.dbver == 2
state = {}
new_ver = reg.apply(state, current_version=0, silent=True)
new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
assert new_ver == 2
assert state["v"] == 2
def test_migrations_can_use_kanta_ctx():
reg = MigrationRegistry()
kanta = _DummyKanta()
@reg.register
def migrate_v1(d, kanta):
kanta.ctx.source = "migration"
d["source"] = kanta.ctx.source
state = {}
new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
assert new_ver == 1
assert state["source"] == "migration"
assert kanta.ctx.source == "migration"
def test_migration_can_omit_kanta_argument():
reg = MigrationRegistry()
kanta = _DummyKanta()
@reg.register
def migrate_v1(d):
d["x"] = 1
state = {}
new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
assert new_ver == 1
assert state["x"] == 1
+153
View File
@@ -0,0 +1,153 @@
"""Tests for Kanta read-only mode."""
import pytest
from kanta.exceptions import DataIntegrityError, FileLockError
from kanta.serialization import struct_to_dict
from .support import (
Data,
EvolvableDataV2,
fixed_change,
make_kanta,
make_migrations_module,
seed_single_change,
)
@pytest.mark.asyncio
async def test_readonly_opens_existing_database(tmp_path, format_config):
path = tmp_path / "test.db"
seed_single_change(path, fixed_change("seed", {"counter": 7}), format_config)
kanta = make_kanta(path, Data, format_config)
await kanta.open(readonly=True)
assert isinstance(kanta.data, Data)
assert kanta.data.counter == 7
assert kanta._impl.readonly is True
assert kanta._impl.background_task is None
await kanta.close()
@pytest.mark.asyncio
async def test_readonly_missing_file_fails(tmp_path, format_config):
path = tmp_path / "missing.db"
kanta = make_kanta(path, Data, format_config)
with pytest.raises(FileLockError):
await kanta.open(readonly=True)
assert not path.exists()
@pytest.mark.asyncio
async def test_readonly_empty_file_fails(tmp_path, format_config):
path = tmp_path / "empty.db"
path.touch()
kanta = make_kanta(path, Data, format_config)
with pytest.raises(DataIntegrityError, match="empty"):
await kanta.open(readonly=True)
@pytest.mark.asyncio
async def test_readonly_transaction_fails(tmp_path, format_config):
path = tmp_path / "test.db"
seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
kanta = make_kanta(path, Data, format_config)
await kanta.open(readonly=True)
with pytest.raises(DataIntegrityError, match="read-only"):
with kanta.transaction(action="inc") as data:
data.counter = 2
# In-memory state must remain unchanged.
assert kanta.data.counter == 1
await kanta.close()
@pytest.mark.asyncio
async def test_readonly_flush_fails(tmp_path, format_config):
path = tmp_path / "test.db"
seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
kanta = make_kanta(path, Data, format_config)
await kanta.open(readonly=True)
with pytest.raises(DataIntegrityError, match="read-only"):
await kanta.flush()
await kanta.close()
@pytest.mark.asyncio
async def test_readonly_create_true_does_not_create_file(tmp_path, format_config):
path = tmp_path / "test.db"
kanta = make_kanta(path, Data, format_config)
with pytest.raises(FileLockError):
await kanta.open(create=True, readonly=True)
assert not path.exists()
@pytest.mark.asyncio
async def test_readonly_does_not_persist_changes(tmp_path, format_config):
path = tmp_path / "test.db"
seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
original_content = path.read_bytes()
kanta = make_kanta(path, Data, format_config)
await kanta.open(readonly=True)
await kanta.close()
assert path.read_bytes() == original_content
@pytest.mark.asyncio
async def test_readonly_runs_migrations(tmp_path, format_config):
path = tmp_path / "test.db"
seed_single_change(
path,
fixed_change("seed", {"counter": 1}, version=0),
format_config,
)
def migrate_v1(data, kanta):
data.setdefault("enabled", True)
migrations = make_migrations_module("readonly_migrations", "migrate_v1", migrate_v1)
kanta = make_kanta(path, EvolvableDataV2, format_config, migrations=migrations)
await kanta.open(readonly=True)
assert kanta.data.counter == 1
# Migration ran in memory even though no change was persisted.
assert struct_to_dict(kanta.data, serializer=kanta._impl.serializer) == {
"counter": 1,
"enabled": True,
}
assert not kanta._impl.pending_changes
await kanta.close()
@pytest.mark.asyncio
async def test_readwrite_and_readonly_can_open_together(tmp_path, format_config):
path = tmp_path / "test.db"
seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
rw = make_kanta(path, Data, format_config)
await rw.open()
ro = make_kanta(path, Data, format_config)
await ro.open(readonly=True)
assert rw.data.counter == 1
assert ro.data.counter == 1
await ro.close()
await rw.close()