Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
bec4635460 | ||
|
|
c04a245366 |
@@ -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:
|
||||||
|
|||||||
@@ -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
|
||||||
|
|
||||||
|
|||||||
+5
-12
@@ -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,11 +318,7 @@ class CallbackRegistry:
|
|||||||
f"Allowed: str path, {self._allowed_message('logfmt')}"
|
f"Allowed: str path, {self._allowed_message('logfmt')}"
|
||||||
)
|
)
|
||||||
|
|
||||||
if sig.return_annotation is inspect.Signature.empty:
|
if sig.return_annotation is not inspect.Signature.empty:
|
||||||
raise TypeError(
|
|
||||||
f"logfmt callback {callback.__name__} must annotate its "
|
|
||||||
f"return type as str | None"
|
|
||||||
)
|
|
||||||
return_ann = self._resolve_raw_annotation(sig.return_annotation, callback)
|
return_ann = self._resolve_raw_annotation(sig.return_annotation, callback)
|
||||||
if not self._is_optional_str(return_ann):
|
if not self._is_optional_str(return_ann):
|
||||||
raise TypeError(
|
raise TypeError(
|
||||||
@@ -416,11 +413,7 @@ class CallbackRegistry:
|
|||||||
f"logfmt class {cls.__name__}.resolve must accept a 'path: str' parameter"
|
f"logfmt class {cls.__name__}.resolve must accept a 'path: str' parameter"
|
||||||
)
|
)
|
||||||
|
|
||||||
if resolve_sig.return_annotation is inspect.Signature.empty:
|
if resolve_sig.return_annotation is not inspect.Signature.empty:
|
||||||
raise TypeError(
|
|
||||||
f"logfmt class {cls.__name__}.resolve must annotate its "
|
|
||||||
f"return type as str | None"
|
|
||||||
)
|
|
||||||
return_ann = self._resolve_raw_annotation(
|
return_ann = self._resolve_raw_annotation(
|
||||||
resolve_sig.return_annotation, resolve
|
resolve_sig.return_annotation, resolve
|
||||||
)
|
)
|
||||||
@@ -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)
|
||||||
|
|||||||
+26
-12
@@ -34,6 +34,7 @@ if sys.platform == "win32":
|
|||||||
_GENERIC_READ = 0x80000000
|
_GENERIC_READ = 0x80000000
|
||||||
_GENERIC_WRITE = 0x40000000
|
_GENERIC_WRITE = 0x40000000
|
||||||
_FILE_SHARE_READ = 0x00000001
|
_FILE_SHARE_READ = 0x00000001
|
||||||
|
_FILE_SHARE_WRITE = 0x00000002
|
||||||
_OPEN_EXISTING = 3
|
_OPEN_EXISTING = 3
|
||||||
_OPEN_ALWAYS = 4
|
_OPEN_ALWAYS = 4
|
||||||
_FILE_ATTRIBUTE_NORMAL = 0x80
|
_FILE_ATTRIBUTE_NORMAL = 0x80
|
||||||
@@ -91,12 +92,13 @@ else:
|
|||||||
|
|
||||||
|
|
||||||
class LockedFile:
|
class LockedFile:
|
||||||
"""A file opened with an exclusive write lock.
|
"""A file opened for read+write with an optional exclusive lock.
|
||||||
|
|
||||||
Usage::
|
Usage::
|
||||||
|
|
||||||
f = LockedFile()
|
f = LockedFile()
|
||||||
f.open(path) # open + lock (read+write)
|
f.open(path) # open + lock (read+write)
|
||||||
|
f.open(path, readonly=True) # open read-only without locking
|
||||||
content = f.read() # read entire content
|
content = f.read() # read entire content
|
||||||
f.write(data) # append data (seeks to end first)
|
f.write(data) # append data (seeks to end first)
|
||||||
f.close() # release lock + close fd
|
f.close() # release lock + close fd
|
||||||
@@ -108,12 +110,13 @@ class LockedFile:
|
|||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
self._fd: int | None = None # Unix fd or Windows HANDLE
|
self._fd: int | None = None # Unix fd or Windows HANDLE
|
||||||
|
|
||||||
def open(self, path: Path, *, create: bool = False) -> None:
|
def open(self, path: Path, *, create: bool = False, readonly: bool = False) -> None:
|
||||||
"""Open *path* for read+write with an exclusive lock.
|
"""Open *path* and optionally acquire an exclusive lock.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
path: File to open and lock.
|
path: File to open and lock.
|
||||||
create: If True, create the file if it doesn't exist (bootstrap).
|
create: If True, create the file if it doesn't exist (bootstrap).
|
||||||
|
readonly: If True, open read-only without acquiring a lock.
|
||||||
|
|
||||||
Raises:
|
Raises:
|
||||||
FileLockError: If the file is locked by another process or not found.
|
FileLockError: If the file is locked by another process or not found.
|
||||||
@@ -122,16 +125,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,12 +191,16 @@ class LockedFile:
|
|||||||
|
|
||||||
# -- Unix ----------------------------------------------------------------
|
# -- Unix ----------------------------------------------------------------
|
||||||
|
|
||||||
def _open_unix(self, path: Path, create: bool) -> None:
|
def _open_unix(self, path: Path, create: bool, readonly: bool) -> None:
|
||||||
|
if readonly:
|
||||||
|
flags = os.O_RDONLY
|
||||||
|
else:
|
||||||
flags = os.O_RDWR | (os.O_CREAT if create else 0)
|
flags = os.O_RDWR | (os.O_CREAT if create else 0)
|
||||||
try:
|
try:
|
||||||
fd = os.open(path, flags, 0o666)
|
fd = os.open(path, flags, 0o666)
|
||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
_fatal(f"Database file not found: {path.resolve()}", db_path=path)
|
_fatal(f"Database file not found: {path.resolve()}", db_path=path)
|
||||||
|
if not readonly:
|
||||||
try:
|
try:
|
||||||
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||||
except OSError:
|
except OSError:
|
||||||
@@ -220,12 +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:
|
||||||
|
if readonly:
|
||||||
|
disposition = _OPEN_EXISTING
|
||||||
|
access = _GENERIC_READ
|
||||||
|
share = _FILE_SHARE_READ | _FILE_SHARE_WRITE
|
||||||
|
else:
|
||||||
disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING
|
disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING
|
||||||
|
access = _GENERIC_READ | _GENERIC_WRITE
|
||||||
|
share = _FILE_SHARE_READ
|
||||||
handle = _kernel32.CreateFileW(
|
handle = _kernel32.CreateFileW(
|
||||||
str(path),
|
str(path),
|
||||||
_GENERIC_READ | _GENERIC_WRITE,
|
access,
|
||||||
_FILE_SHARE_READ,
|
share,
|
||||||
None,
|
None,
|
||||||
disposition,
|
disposition,
|
||||||
_FILE_ATTRIBUTE_NORMAL,
|
_FILE_ATTRIBUTE_NORMAL,
|
||||||
|
|||||||
+15
-6
@@ -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.
|
||||||
|
|||||||
+23
-4
@@ -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)
|
||||||
|
if self.readonly:
|
||||||
|
self.statedict = copy.deepcopy(normalized)
|
||||||
|
else:
|
||||||
self.queue_change("migrate:msgspec", normalized, mtime=False)
|
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,6 +196,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
|||||||
|
|
||||||
self.opened = True
|
self.opened = True
|
||||||
|
|
||||||
|
if not self.readonly:
|
||||||
self.background_task = asyncio.create_task(self._background_loop())
|
self.background_task = asyncio.create_task(self._background_loop())
|
||||||
|
|
||||||
async def close(self) -> None:
|
async def close(self) -> None:
|
||||||
@@ -196,6 +214,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
|||||||
|
|
||||||
# Always run a final flush in case the background task never reached
|
# Always run a final flush in case the background task never reached
|
||||||
# its cancellation handler.
|
# its cancellation handler.
|
||||||
|
if not self.readonly:
|
||||||
await self.flush()
|
await self.flush()
|
||||||
|
|
||||||
self.file.close()
|
self.file.close()
|
||||||
|
|||||||
+21
-16
@@ -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(".")
|
||||||
|
|||||||
@@ -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
|
||||||
|
|
||||||
|
|||||||
@@ -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 "
|
||||||
|
|||||||
+48
-4
@@ -1,4 +1,4 @@
|
|||||||
from typing import Any
|
from typing import Any, Optional, Union
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
@@ -49,16 +49,60 @@ def test_logfmt_requires_value_annotation(tmp_path, format_config):
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
def test_logfmt_requires_return_annotation(tmp_path, format_config):
|
def test_logfmt_allows_missing_return_annotation(tmp_path, format_config):
|
||||||
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
||||||
|
|
||||||
with pytest.raises(TypeError, match="must annotate its return"):
|
|
||||||
|
|
||||||
@kanta.logfmt
|
@kanta.logfmt
|
||||||
def resolve_names(value: str, current: DictPost):
|
def resolve_names(value: str, current: DictPost):
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def test_logfmt_class_allows_missing_return_annotation(tmp_path, format_config):
|
||||||
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
||||||
|
|
||||||
|
@kanta.logfmt
|
||||||
|
class UserLogFmt(LogFmt):
|
||||||
|
def resolve(self, value: str, path: str):
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
# fmt: off
|
||||||
|
def test_logfmt_accepts_optional_return_typing_forms(tmp_path, format_config):
|
||||||
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
||||||
|
|
||||||
|
@kanta.logfmt
|
||||||
|
def resolve_optional(value: str) -> Optional[str]: # noqa: UP007
|
||||||
|
return value
|
||||||
|
|
||||||
|
@kanta.logfmt
|
||||||
|
def resolve_union(value: str) -> Union[str, None]: # noqa: UP007
|
||||||
|
return value
|
||||||
|
|
||||||
|
@kanta.logfmt
|
||||||
|
def resolve_pipe(value: "str") -> "str | None":
|
||||||
|
return value
|
||||||
|
|
||||||
|
|
||||||
|
def test_logfmt_class_accepts_optional_return_typing_forms(tmp_path, format_config):
|
||||||
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
||||||
|
|
||||||
|
@kanta.logfmt
|
||||||
|
class OptionalStyle(LogFmt):
|
||||||
|
def resolve(self, value: str, path: str) -> Optional[str]: # noqa: UP007
|
||||||
|
return value
|
||||||
|
|
||||||
|
@kanta.logfmt
|
||||||
|
class UnionStyle(LogFmt):
|
||||||
|
def resolve(self, value: str, path: str) -> Union[str, None]: # noqa: UP007
|
||||||
|
return value
|
||||||
|
|
||||||
|
@kanta.logfmt
|
||||||
|
class StringStyle(LogFmt):
|
||||||
|
def resolve(self, value: "str", path: "str") -> "str | None":
|
||||||
|
return value
|
||||||
|
# fmt: on
|
||||||
|
|
||||||
|
|
||||||
def test_logfmt_rejects_async_callback(tmp_path, format_config):
|
def test_logfmt_rejects_async_callback(tmp_path, format_config):
|
||||||
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
kanta = make_kanta(tmp_path / "test.db", Data, format_config)
|
||||||
|
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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()
|
||||||
Reference in New Issue
Block a user