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 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:
+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()` (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
+18 -25
View File
@@ -12,6 +12,7 @@ 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
@@ -317,17 +318,13 @@ 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( return_ann = self._resolve_raw_annotation(sig.return_annotation, callback)
f"logfmt callback {callback.__name__} must annotate its " if not self._is_optional_str(return_ann):
f"return type as str | None" raise TypeError(
) f"logfmt callback {callback.__name__} must return str | None, "
return_ann = self._resolve_raw_annotation(sig.return_annotation, callback) f"got {return_ann!r}"
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( return _LogFmtFunctionSpec(
callback=callback, callback=callback,
@@ -416,19 +413,15 @@ 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( return_ann = self._resolve_raw_annotation(
f"logfmt class {cls.__name__}.resolve must annotate its " resolve_sig.return_annotation, resolve
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 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) return _LogFmtClassSpec(cls=cls, inject_params=inject_params, path=path)
@@ -516,7 +509,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 +517,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)
+40 -26
View File
@@ -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,15 +92,16 @@ 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)
content = f.read() # read entire content f.open(path, readonly=True) # open read-only without locking
f.write(data) # append data (seeks to end first) content = f.read() # read entire content
f.close() # release lock + close fd 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. Unix: fcntl.flock (advisory) — read-only callers that don't flock are unaffected.
Windows: CreateFileW with FILE_SHARE_READ — OS blocks other writers. Windows: CreateFileW with FILE_SHARE_READ — OS blocks other writers.
@@ -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,16 @@ 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(self, path: Path, create: bool = False, readonly: bool = False) -> bytes:
"""Open *path* with exclusive lock and read all content. """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,20 +191,24 @@ class LockedFile:
# -- Unix ---------------------------------------------------------------- # -- Unix ----------------------------------------------------------------
def _open_unix(self, path: Path, create: bool) -> None: def _open_unix(self, path: Path, create: bool, readonly: bool) -> None:
flags = os.O_RDWR | (os.O_CREAT if create else 0) if readonly:
flags = os.O_RDONLY
else:
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)
try: if not readonly:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) try:
except OSError: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
os.close(fd) except OSError:
_fatal( os.close(fd)
f"{path.resolve()}: database already locked by another instance", _fatal(
db_path=path, f"{path.resolve()}: database already locked by another instance",
) db_path=path,
)
self._fd = fd self._fd = fd
def _read_unix(self) -> bytes: def _read_unix(self) -> bytes:
@@ -220,12 +227,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:
disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING 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( 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,
+15 -6
View File
@@ -3,7 +3,7 @@
from __future__ import annotations from __future__ import annotations
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 Any, Generic, TypeVar
from kanta.kantaimpl import KantaImpl from kanta.kantaimpl import KantaImpl
@@ -50,7 +50,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 +60,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 +76,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 +124,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 +144,7 @@ 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) -> 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 +153,9 @@ 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.
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 +163,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)
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.
+26 -7
View File
@@ -7,6 +7,7 @@ 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
@@ -28,8 +29,8 @@ class KantaImpl(PersistenceMixin, Generic[T]):
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.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)
self.ctx = SimpleNamespace()
super().__init__(**kwargs) super().__init__(**kwargs)
self.migration_registry: MigrationRegistry | None = None self.migration_registry: MigrationRegistry | None = None
if self.migrations is not None: if self.migrations is not None:
@@ -43,6 +44,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
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
@@ -75,7 +77,7 @@ 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: async def open(self, *, create: bool = True, readonly: bool = False) -> 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 +86,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):
@@ -136,7 +143,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
if self.migration_registry is not None: if self.migration_registry is not None:
rr.version = self.migration_registry.apply( 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) self.statedict = copy.deepcopy(rr.state)
@@ -149,12 +156,22 @@ class KantaImpl(PersistenceMixin, Generic[T]):
self.version = rr.version self.version = rr.version
self.mtime = rr.m self.mtime = rr.m
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.statedict = copy.deepcopy(normalized)
else:
self.queue_change("migrate:msgspec", normalized, mtime=False)
self.snapshot.ts = ( self.snapshot.ts = (
datetime.fromtimestamp(rr.last_snapshot_mtime, UTC) datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)
if rr.last_snapshot_mtime is not None if rr.last_snapshot_mtime is not None
else 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"): elif self.callback_registry.has("bootstrap"):
try: try:
await self.callback_registry.invoke( await self.callback_registry.invoke(
@@ -179,7 +196,8 @@ class KantaImpl(PersistenceMixin, Generic[T]):
self.opened = True 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: async def close(self) -> None:
"""Stop the background task, flush pending changes, and release the file lock.""" """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 # Always run a final flush in case the background task never reached
# its cancellation handler. # its cancellation handler.
await self.flush() if not self.readonly:
await self.flush()
self.file.close() self.file.close()
self.opened = False 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 from __future__ import annotations
import importlib import importlib
import inspect
import logging import logging
from types import ModuleType from types import ModuleType
from typing import Any from typing import Any
import msgspec
_logger = logging.getLogger(__name__) _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: class MigrationRegistry:
"""Registry of schema migration functions. """Registry of schema migration functions.
@@ -33,15 +23,20 @@ class MigrationRegistry:
registry = MigrationRegistry() registry = MigrationRegistry()
@registry.register @registry.register
def migrate_v1(d: dict, ctx: MigrationCtx) -> None: def migrate_v1(d: dict, kanta) -> None:
d.setdefault("version", 1) 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:: Or load from a module::
registry = MigrationRegistry.from_module("myapp.migrations") 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: def __init__(self) -> None:
@@ -89,11 +84,21 @@ class MigrationRegistry:
"""Current schema version (= highest discovered migration, or 0).""" """Current schema version (= highest discovered migration, or 0)."""
return max(self._migrations.keys(), default=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( def apply(
self, self,
data_dict: dict[str, Any], data_dict: dict[str, Any],
current_version: int, current_version: int,
ctx: MigrationCtx | None = None, kanta: Any,
*, *,
silent: bool = False, silent: bool = False,
) -> int: ) -> int:
@@ -109,7 +114,7 @@ class MigrationRegistry:
f"Missing migration step migrate_v{next_version} " f"Missing migration step migrate_v{next_version} "
f"(highest discovered is v{self.dbver})" f"(highest discovered is v{self.dbver})"
) )
fn(data_dict, ctx or MigrationCtx()) self._call_migration(fn, data_dict, kanta)
current_version = next_version current_version = next_version
if not silent: if not silent:
desc = (fn.__doc__ or fn.__name__).split("\n")[0].rstrip(".") desc = (fn.__doc__ or fn.__name__).split("\n")[0].rstrip(".")
+17
View File
@@ -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)
@@ -160,6 +163,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 +217,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
+7
View File
@@ -24,6 +24,13 @@ def transaction(
mtime: bool | datetime = True, mtime: bool | datetime = 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 "
+49 -5
View File
@@ -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
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 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)
+2 -2
View File
@@ -404,7 +404,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
@@ -514,7 +514,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
+47 -9
View File
@@ -1,44 +1,52 @@
from types import ModuleType from types import ModuleType, SimpleNamespace
from kanta.migrate import MigrationRegistry from kanta.migrate import MigrationRegistry
class _DummyKanta:
def __init__(self):
self.ctx = SimpleNamespace()
def test_register_and_apply(): def test_register_and_apply():
reg = MigrationRegistry() reg = MigrationRegistry()
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) new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
assert new_ver == 2 assert new_ver == 2
assert state["version"] == 2 assert state["version"] == 2
def test_no_migrations_needed(): def test_no_migrations_needed():
reg = MigrationRegistry() reg = MigrationRegistry()
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) new_ver = reg.apply(state, current_version=1, kanta=kanta, silent=True)
assert new_ver == 1 assert new_ver == 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
@@ -48,6 +56,36 @@ def test_from_module():
assert reg.dbver == 2 assert reg.dbver == 2
state = {} 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 new_ver == 2
assert state["v"] == 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()