From d18ebf96a110c98be616aca771f040a504b28a70 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Wed, 21 Oct 2026 12:31:42 -0700 Subject: [PATCH 1/6] Add Cache.lock() atomic locks for array, file and database stores --- docs/adr/031-cache-locks.md | 56 +++++ docs/adr/AGENTS.md | 1 + .../src/fastapi_startkit/cache/__init__.py | 3 + .../src/fastapi_startkit/cache/array_store.py | 32 +++ .../fastapi_startkit/cache/database_store.py | 37 +++- .../src/fastapi_startkit/cache/file_store.py | 59 ++++- .../src/fastapi_startkit/cache/lock.py | 65 ++++++ .../src/fastapi_startkit/cache/repository.py | 11 +- .../src/fastapi_startkit/facades/Cache.pyi | 3 +- fastapi_startkit/tests/cache/test_lock.py | 202 ++++++++++++++++++ 10 files changed, 462 insertions(+), 7 deletions(-) create mode 100644 docs/adr/031-cache-locks.md create mode 100644 fastapi_startkit/src/fastapi_startkit/cache/lock.py create mode 100644 fastapi_startkit/tests/cache/test_lock.py diff --git a/docs/adr/031-cache-locks.md b/docs/adr/031-cache-locks.md new file mode 100644 index 00000000..7cf216c3 --- /dev/null +++ b/docs/adr/031-cache-locks.md @@ -0,0 +1,56 @@ +--- +title: Atomic cache locks with Cache.lock() +date: 2026-10-21 +--- + +## Problem + +Applications need to make sure only one worker runs a piece of work at a time (rebuilding a report, charging an order, running a scheduled job on one of many servers). Laravel offers `Cache::lock($name, $seconds)` with `get`, `block`, `release` and `forceRelease`, backed by the cache store so every worker that shares the store sees the same lock. The cache module has no locks, and the check-then-write `put` cannot be used for one because two workers can both see the key missing. + +## Alternatives + +- **Lock built on `get` + `put`.** Not atomic across processes. Rejected. +- **`asyncio.Lock`.** Process-local only. Rejected. +- **One generic lock that calls a new atomic `add(key, value, seconds)` on every store.** Each store still needs its own atomic primitive, and release must compare the owner and delete atomically too, which `add` does not cover. Rejected. +- **A `Lock` base class with the shared `get`/`block`/`async with` behaviour, and one subclass per store implementing `acquire`, `release` and `force_release` with that store's atomic primitive.** Chosen, matching Laravel's `Lock` / `CacheLock` / `DatabaseLock` split. + +## Decision + +- `cache/lock.py` defines `LockTimeoutError` and the abstract `Lock(name, seconds=0, owner=None)`. `owner` defaults to a random token; passing a known owner restores a lock acquired elsewhere (another request or process) so it can be released there. `seconds <= 0` means the lock never expires. + - `await lock.get()` tries once and returns `bool`; `await lock.get(callback)` runs the sync or async callback only when acquired, releases afterwards (also on error) and returns the callback result, or `False` when not acquired. + - `await lock.block(seconds, callback=None)` retries every 250 ms (`lock.sleep`) until acquired or `seconds` passed, then raises `LockTimeoutError`; with a callback it runs and releases like `get`. + - `async with Cache.lock("name", 10):` acquires once or raises `LockTimeoutError`, and releases on exit. Use `block` to wait. + - `release()` deletes the lock only when this owner holds it and returns `bool`; `force_release()` deletes it regardless of owner. + - Locks are not re-entrant: acquiring a lock the same owner already holds fails, on every store. +- Store locks: + - `ArrayStore`: a `locks` dict of `name -> (owner, expires_at)` on the store. Check and set run without an `await`, so they are atomic within the event loop. + - `FileStore`: one file per lock under `/locks/`, holding JSON `[owner, expires_at]`. Each operation opens the file, takes an exclusive `fcntl.flock`, reads, decides, rewrites or truncates it, and closes it, so acquire and owner-checked release are atomic across processes on one host. Files are truncated, never unlinked, because unlinking a file another process has open would let both lock different inodes. Lock files live outside the two-hex shard directories, so `flush` does not touch them. `fcntl` is imported on use, so the file cache still imports on Windows; locks on the file store are POSIX-only. + - `DatabaseStore`: a row in the cache table with key `lock:`, the pickled owner as value and the usual expiration column. Acquire inserts; on a primary-key conflict it updates the row only where `expiration <= now`, and the lock is acquired when one row changed. Release deletes where key and value match the owner. Reusing the cache table needs no new migration; `flush` therefore also clears database locks. The PostgreSQL transaction caveat of ADR 029 applies to the insert conflict. + - Redis follows in a separate PR on top of the Redis store (`SET NX EX` and an owner-checked delete script). +- `Repository.lock(name, seconds=0, owner=None)` prefixes the name like a cache key and returns the store's lock; a store without locks raises `TypeError`. The facade reaches it through the manager; the stub adds `lock`. + +## Usage + +```python +from fastapi_startkit.facades import Cache + +lock = Cache.lock("reports", 10) +if await lock.get(): + try: + await build_report() + finally: + await lock.release() + +await Cache.lock("reports", 10).get(build_report) +await Cache.lock("reports", 10).block(5, build_report) + +async with Cache.lock("reports", 10): + await build_report() + +owner = lock.owner +await Cache.lock("reports", 10, owner).release() +``` + +## Validation + +`tests/cache/test_lock.py` (unittest) runs one shared case class against the array, file and database (SQLite) stores: acquire once, second owner fails, same owner fails, expired lock can be taken, `seconds=0` never expires, release only by owner, force release, restore by owner, `get` with sync/async callback releases also on error, `block` waits for a release and times out with `LockTimeoutError`, `async with` acquires/releases and raises when held, the repository prefixes lock names, and a store without locks raises `TypeError`. The file store also checks that `flush` keeps locks and that a corrupt lock file counts as free. diff --git a/docs/adr/AGENTS.md b/docs/adr/AGENTS.md index fb3492d8..a9b24b8e 100644 --- a/docs/adr/AGENTS.md +++ b/docs/adr/AGENTS.md @@ -33,3 +33,4 @@ Read this index first, then open the records relevant to the change. Before impl | [025](025-model-lifecycle-events.md) | 2026-10-20 | Model lifecycle events, automatic timestamps and datetime casts | `save()` fires `saving → creating/updating → query → created/updated → saved`. Observer changes are persisted, and `updating` fires only when the model is dirty. A new `Model.delete()` fires `deleting → DELETE → deleted`. `observe_events` is async and awaits async handlers. Mass query-builder writes fire no events. Laravel-style auto timestamps: `created_at`/`updated_at` are filled on create unless supplied, `updated_at` moves on update, and `__timestamps__ = False` disables both. `DateCast.set` returns a UTC datetime; `Connection.prepare_binding` passes it unchanged to Postgres/MySQL, and SQLite converts it to a naive UTC string. Postgres registers an asyncpg `timestamp` codec that stores aware values as naive UTC; `timestamptz` keeps asyncpg's codec. `Annotated[T, Cast]` (a `BaseCast` subclass or instance) overrides a column's cast, and `Attribute.__casts__` is removed. | | [029](029-cache-module.md) | 2026-10-21 | Cache module with callable remember and pluggable stores | New `fastapi_startkit.cache`: async `Cache.get/put/has/forget/flush`, `get` with a callable default, and `remember(key, ttl, factory)` that runs a sync or async factory only on a miss; `remember(key, ttl)` without a factory is a decorator keyed by `key` plus a hash of the call arguments. Stores implement an async contract with a `MISSING` sentinel (cached `None` is a hit): array, file (pickled per-key files) and database (masoniteorm table). TTL in seconds or `timedelta`, `None` forever, <=0 forgets. `CacheConfig` (`CACHE_CONNECTION`, `CACHE_PREFIX`, `stores`) via `resolve_config`; `CacheProvider` binds `cache`; facade `remember` resolves lazily. A `with` form is impossible in Python. Shipped in slices into `feature/caches`. | | [030](030-cache-memo.md) | 2026-10-21 | Request-scoped cache memoization with Cache.memo() | `Cache.memo(name=None)` returns a `Repository` over `MemoStore`, which wraps the named store and memoizes reads (misses too) for the current asyncio task; `put`/`forget` write through and drop the key, `flush` clears memo and store. State is a `ContextVar` holding `(weakref to owning task, memo)`, and a task that finds another task's memo starts a fresh one, so requests, jobs and child tasks never share memo state. Memoized values ignore TTL for the rest of the task. | +| [031](031-cache-locks.md) | 2026-10-21 | Atomic cache locks with Cache.lock() | `Cache.lock(name, seconds=0, owner=None)` returns a store-specific `Lock` (abstract base in `cache/lock.py`): `get(callback=None)` tries once, `block(seconds, callback=None)` polls every 250 ms then raises `LockTimeoutError`, `async with` acquires once or raises, `release()` is owner-checked, `force_release()` is not; callbacks release on error; passing `owner` restores a lock elsewhere; not re-entrant; `seconds<=0` never expires. Array: dict on the store, atomic in the event loop. File: `/locks/` JSON `[owner, expires_at]` under `fcntl.flock`, truncated never unlinked, kept by `flush`, POSIX-only. Database: row `lock:` in the cache table, insert then update-if-expired on conflict, release deletes by owner; no new migration, `flush` clears it. Redis lock follows on the Redis store. | diff --git a/fastapi_startkit/src/fastapi_startkit/cache/__init__.py b/fastapi_startkit/src/fastapi_startkit/cache/__init__.py index f11d6179..c55bc155 100644 --- a/fastapi_startkit/src/fastapi_startkit/cache/__init__.py +++ b/fastapi_startkit/src/fastapi_startkit/cache/__init__.py @@ -1,6 +1,7 @@ from .array_store import ArrayStore from .config import ArrayStoreConfig, CacheConfig, DatabaseStoreConfig, FileStoreConfig from .file_store import FileStore +from .lock import Lock, LockTimeoutError from .manager import CacheManager from .memo_store import MemoStore from .provider import CacheProvider @@ -17,6 +18,8 @@ "DatabaseStoreConfig", "FileStore", "FileStoreConfig", + "Lock", + "LockTimeoutError", "MemoStore", "Repository", "Store", diff --git a/fastapi_startkit/src/fastapi_startkit/cache/array_store.py b/fastapi_startkit/src/fastapi_startkit/cache/array_store.py index 91698ff3..05a85338 100644 --- a/fastapi_startkit/src/fastapi_startkit/cache/array_store.py +++ b/fastapi_startkit/src/fastapi_startkit/cache/array_store.py @@ -1,11 +1,13 @@ from typing import Any +from .lock import Lock from .store import MISSING, expired, expires_at class ArrayStore: def __init__(self) -> None: self.storage: dict[str, tuple[float | None, Any]] = {} + self.locks: dict[str, tuple[str, float | None]] = {} async def get(self, key: str) -> Any: entry = self.storage.get(key) @@ -28,3 +30,33 @@ async def forget(self, key: str) -> bool: async def flush(self) -> bool: self.storage.clear() return True + + def lock(self, name: str, seconds: float = 0, owner: str | None = None) -> "ArrayLock": + return ArrayLock(self, name, seconds, owner) + + +class ArrayLock(Lock): + def __init__(self, store: ArrayStore, name: str, seconds: float = 0, owner: str | None = None): + super().__init__(name, seconds, owner) + self.store = store + + def holder(self) -> str | None: + entry = self.store.locks.get(self.name) + if entry is None or expired(entry[1]): + return None + return entry[0] + + async def acquire(self) -> bool: + if self.holder() is not None: + return False + self.store.locks[self.name] = (self.owner, expires_at(self.ttl)) + return True + + async def release(self) -> bool: + if self.holder() != self.owner: + return False + del self.store.locks[self.name] + return True + + async def force_release(self) -> None: + self.store.locks.pop(self.name, None) diff --git a/fastapi_startkit/src/fastapi_startkit/cache/database_store.py b/fastapi_startkit/src/fastapi_startkit/cache/database_store.py index 2aad1893..a10e118e 100644 --- a/fastapi_startkit/src/fastapi_startkit/cache/database_store.py +++ b/fastapi_startkit/src/fastapi_startkit/cache/database_store.py @@ -1,6 +1,7 @@ import base64 import math import pickle +import time from collections.abc import Callable from typing import Any @@ -9,9 +10,14 @@ from fastapi_startkit.masoniteorm.connections.connection import Connection from fastapi_startkit.masoniteorm.models.builder import QueryBuilder +from .lock import Lock from .store import MISSING, expired, expires_at +def encode(value: Any) -> str: + return base64.b64encode(pickle.dumps(value)).decode() + + class DatabaseStore: def __init__(self, connection: Callable[[], Connection], table: str = "caches"): self.connection = connection @@ -39,10 +45,7 @@ async def get(self, key: str) -> Any: async def put(self, key: str, value: Any, seconds: float | None) -> bool: expiration = expires_at(seconds) - record = { - "value": base64.b64encode(pickle.dumps(value)).decode(), - "expiration": None if expiration is None else math.ceil(expiration), - } + record = {"value": encode(value), "expiration": None if expiration is None else math.ceil(expiration)} if await self.query().where("key", key).update(record): return True try: @@ -57,3 +60,29 @@ async def forget(self, key: str) -> bool: async def flush(self) -> bool: await self.query().delete() return True + + def lock(self, name: str, seconds: float = 0, owner: str | None = None) -> "DatabaseLock": + return DatabaseLock(self, name, seconds, owner) + + +class DatabaseLock(Lock): + def __init__(self, store: DatabaseStore, name: str, seconds: float = 0, owner: str | None = None): + super().__init__(name, seconds, owner) + self.store = store + self.key = f"lock:{name}" + + async def acquire(self) -> bool: + expiration = expires_at(self.ttl) + record = {"value": encode(self.owner), "expiration": None if expiration is None else math.ceil(expiration)} + try: + await self.store.query().insert({"key": self.key, **record}) + except IntegrityError: + expired_lock = self.store.query().where("key", self.key).where("expiration", "<=", time.time()) + return bool(await expired_lock.update(record)) + return True + + async def release(self) -> bool: + return bool(await self.store.query().where("key", self.key).where("value", encode(self.owner)).delete()) + + async def force_release(self) -> None: + await self.store.query().where("key", self.key).delete() diff --git a/fastapi_startkit/src/fastapi_startkit/cache/file_store.py b/fastapi_startkit/src/fastapi_startkit/cache/file_store.py index d4777431..fda1e33e 100644 --- a/fastapi_startkit/src/fastapi_startkit/cache/file_store.py +++ b/fastapi_startkit/src/fastapi_startkit/cache/file_store.py @@ -1,16 +1,21 @@ import asyncio import hashlib +import json import os import pickle import re import shutil import tempfile +from collections.abc import Callable from pathlib import Path from typing import Any -from .store import MISSING, expired, expires_at +from .lock import Lock +from .store import MISSING, Missing, expired, expires_at SHARD = re.compile(r"^[0-9a-f]{2}$") +Holder = tuple[str, float | None] | None +Decision = tuple[bool, Holder | Missing] class FileStore: @@ -76,3 +81,55 @@ def clear(self) -> None: for shard in self.directory.iterdir(): if shard.is_dir() and SHARD.match(shard.name): shutil.rmtree(shard, ignore_errors=True) + + def lock(self, name: str, seconds: float = 0, owner: str | None = None) -> "FileLock": + return FileLock(self, name, seconds, owner) + + +class FileLock(Lock): + def __init__(self, store: FileStore, name: str, seconds: float = 0, owner: str | None = None): + super().__init__(name, seconds, owner) + self.path = store.directory / "locks" / hashlib.sha1(name.encode()).hexdigest() + + async def acquire(self) -> bool: + def decide(holder: Holder) -> Decision: + if holder is not None: + return False, MISSING + return True, (self.owner, expires_at(self.ttl)) + + return await asyncio.to_thread(self.exclusive, decide) + + async def release(self) -> bool: + def decide(holder: Holder) -> Decision: + if holder is None or holder[0] != self.owner: + return False, MISSING + return True, None + + return await asyncio.to_thread(self.exclusive, decide) + + async def force_release(self) -> None: + await asyncio.to_thread(self.exclusive, lambda holder: (True, None)) + + def exclusive(self, decide: Callable[[Holder], Decision]) -> bool: + import fcntl + + self.path.parent.mkdir(parents=True, exist_ok=True) + with open(self.path, "a+b") as file: + fcntl.flock(file, fcntl.LOCK_EX) + file.seek(0) + result, holder = decide(self.read(file.read())) + if holder is not MISSING: + file.truncate(0) + if holder is not None: + file.write(json.dumps(holder).encode()) + file.flush() + return result + + def read(self, payload: bytes) -> Holder: + try: + owner, expiration = json.loads(payload) + except (ValueError, TypeError): + return None + if not isinstance(owner, str) or not isinstance(expiration, int | float | None) or expired(expiration): + return None + return owner, expiration diff --git a/fastapi_startkit/src/fastapi_startkit/cache/lock.py b/fastapi_startkit/src/fastapi_startkit/cache/lock.py new file mode 100644 index 00000000..d8252b52 --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/cache/lock.py @@ -0,0 +1,65 @@ +import asyncio +import secrets +import time +from abc import ABC, abstractmethod +from collections.abc import Callable +from typing import Any + +from .repository import evaluate + + +class LockTimeoutError(Exception): + pass + + +class Lock(ABC): + sleep: float = 0.25 + + def __init__(self, name: str, seconds: float = 0, owner: str | None = None): + self.name = name + self.seconds = seconds + self.owner = owner or secrets.token_hex(16) + + @property + def ttl(self) -> float | None: + return self.seconds if self.seconds > 0 else None + + @abstractmethod + async def acquire(self) -> bool: ... + + @abstractmethod + async def release(self) -> bool: ... + + @abstractmethod + async def force_release(self) -> None: ... + + async def get(self, callback: Callable[[], Any] | None = None) -> Any: + if not await self.acquire(): + return False + if callback is None: + return True + return await self.run(callback) + + async def block(self, seconds: float, callback: Callable[[], Any] | None = None) -> Any: + deadline = time.monotonic() + seconds + while not await self.acquire(): + if time.monotonic() >= deadline: + raise LockTimeoutError(f"Could not acquire lock [{self.name}] within {seconds} seconds.") + await asyncio.sleep(self.sleep) + if callback is None: + return True + return await self.run(callback) + + async def run(self, callback: Callable[[], Any]) -> Any: + try: + return await evaluate(callback) + finally: + await self.release() + + async def __aenter__(self) -> "Lock": + if not await self.acquire(): + raise LockTimeoutError(f"Lock [{self.name}] is held by another owner.") + return self + + async def __aexit__(self, *exc: object) -> None: + await self.release() diff --git a/fastapi_startkit/src/fastapi_startkit/cache/repository.py b/fastapi_startkit/src/fastapi_startkit/cache/repository.py index a5f6630e..115414a6 100644 --- a/fastapi_startkit/src/fastapi_startkit/cache/repository.py +++ b/fastapi_startkit/src/fastapi_startkit/cache/repository.py @@ -4,10 +4,13 @@ import pickle from collections.abc import Awaitable, Callable from datetime import timedelta -from typing import Any, ParamSpec, Protocol, overload +from typing import TYPE_CHECKING, Any, ParamSpec, Protocol, overload from .store import MISSING, Store +if TYPE_CHECKING: + from .lock import Lock + P = ParamSpec("P") Ttl = float | timedelta | None @@ -106,3 +109,9 @@ async def forget(self, key: str) -> bool: async def flush(self) -> bool: return await self.store.flush() + + def lock(self, name: str, seconds: float = 0, owner: str | None = None) -> "Lock": + factory = getattr(self.store, "lock", None) + if factory is None: + raise TypeError(f"Cache store [{type(self.store).__name__}] does not support locks.") + return factory(self.key(name), seconds, owner) diff --git a/fastapi_startkit/src/fastapi_startkit/facades/Cache.pyi b/fastapi_startkit/src/fastapi_startkit/facades/Cache.pyi index 2e0c2ffa..6264dcb3 100644 --- a/fastapi_startkit/src/fastapi_startkit/facades/Cache.pyi +++ b/fastapi_startkit/src/fastapi_startkit/facades/Cache.pyi @@ -1,7 +1,7 @@ from collections.abc import Awaitable, Callable from typing import Any, overload -from fastapi_startkit.cache import CacheManager, Repository +from fastapi_startkit.cache import CacheManager, Lock, Repository from fastapi_startkit.cache.manager import StoreFactory from fastapi_startkit.cache.repository import Decorator, Ttl @@ -17,6 +17,7 @@ class CacheFacade: def remember(self, key: str, ttl: Ttl, factory: Callable[[], Any]) -> Awaitable[Any]: ... @overload def remember(self, key: str | None, ttl: Ttl, factory: None = None) -> Decorator: ... + def lock(self, name: str, seconds: float = 0, owner: str | None = None) -> Lock: ... async def forget(self, key: str) -> bool: ... async def flush(self) -> bool: ... diff --git a/fastapi_startkit/tests/cache/test_lock.py b/fastapi_startkit/tests/cache/test_lock.py new file mode 100644 index 00000000..6aad7459 --- /dev/null +++ b/fastapi_startkit/tests/cache/test_lock.py @@ -0,0 +1,202 @@ +import asyncio +import tempfile +from pathlib import Path +from unittest import IsolatedAsyncioTestCase, mock + +from fastapi_startkit.cache import ArrayStore, FileStore, LockTimeoutError, Repository +from fastapi_startkit.cache.database_store import DatabaseStore + +from test_database_store import InMemoryDatabase, load_migration + + +class LockContract: + async def make_store(self): + raise NotImplementedError + + async def asyncSetUp(self): + self.time = self.enterContext(mock.patch("time.time", return_value=1_000_000.0)) + self.store = await self.make_store() + + def lock(self, seconds=10, owner=None): + lock = self.store.lock("reports", seconds, owner) + lock.sleep = 0.01 + return lock + + async def test_acquires_once(self): + self.assertIs(await self.lock().get(), True) + self.assertIs(await self.lock().get(), False) + + async def test_is_not_reentrant(self): + lock = self.lock() + + self.assertIs(await lock.get(), True) + self.assertIs(await lock.get(), False) + + async def test_expired_lock_can_be_taken(self): + await self.lock(seconds=10).get() + + self.time.return_value += 11 + + self.assertIs(await self.lock().get(), True) + + async def test_zero_seconds_never_expires(self): + await self.lock(seconds=0).get() + + self.time.return_value += 10**9 + + self.assertIs(await self.lock().get(), False) + + async def test_release_only_by_owner(self): + lock = self.lock() + await lock.get() + + self.assertIs(await self.lock().release(), False) + self.assertIs(await self.lock().get(), False) + self.assertIs(await lock.release(), True) + self.assertIs(await lock.release(), False) + self.assertIs(await self.lock().get(), True) + + async def test_force_release_ignores_owner(self): + await self.lock().get() + + await self.lock().force_release() + + self.assertIs(await self.lock().get(), True) + + async def test_restored_owner_can_release(self): + lock = self.lock() + await lock.get() + + self.assertIs(await self.lock(owner=lock.owner).release(), True) + self.assertIs(await self.lock().get(), True) + + async def test_get_runs_sync_callback_and_releases(self): + self.assertEqual(await self.lock().get(lambda: "done"), "done") + self.assertIs(await self.lock().get(), True) + + async def test_get_runs_async_callback_and_releases(self): + async def build(): + return "done" + + self.assertEqual(await self.lock().get(build), "done") + self.assertIs(await self.lock().get(), True) + + async def test_get_skips_callback_when_held(self): + callback = mock.Mock() + await self.lock().get() + + self.assertIs(await self.lock().get(callback), False) + callback.assert_not_called() + + async def test_get_releases_when_callback_fails(self): + def fail(): + raise RuntimeError("boom") + + with self.assertRaises(RuntimeError): + await self.lock().get(fail) + + self.assertIs(await self.lock().get(), True) + + async def test_block_waits_for_release(self): + holder = self.lock() + await holder.get() + waits = [] + sleep = asyncio.sleep + + async def release_while_waiting(seconds): + waits.append(seconds) + await holder.release() + await sleep(0) + + with mock.patch("asyncio.sleep", release_while_waiting): + self.assertEqual(await self.lock().block(2, lambda: "done"), "done") + + self.assertEqual(waits, [0.01]) + self.assertIs(await self.lock().get(), True) + + async def test_block_times_out(self): + await self.lock().get() + + with self.assertRaises(LockTimeoutError): + await self.lock().block(0.05) + + async def test_block_returns_true_without_callback(self): + self.assertIs(await self.lock().block(1), True) + + async def test_async_with_acquires_and_releases(self): + async with self.lock(): + self.assertIs(await self.lock().get(), False) + + self.assertIs(await self.lock().get(), True) + + async def test_async_with_raises_when_held(self): + await self.lock().get() + + with self.assertRaises(LockTimeoutError): + async with self.lock(): + self.fail("body must not run") + + +class TestArrayLock(LockContract, IsolatedAsyncioTestCase): + async def make_store(self): + return ArrayStore() + + +class TestFileLock(LockContract, IsolatedAsyncioTestCase): + async def make_store(self): + directory = self.enterContext(tempfile.TemporaryDirectory()) + return FileStore(directory) + + async def test_flush_keeps_locks(self): + await self.lock().get() + + await self.store.flush() + + self.assertIs(await self.lock().get(), False) + + async def test_corrupt_lock_file_counts_as_free(self): + lock = self.lock() + lock.path.parent.mkdir(parents=True) + lock.path.write_bytes(b"not json") + + self.assertIs(await lock.get(), True) + + async def test_release_truncates_instead_of_unlinking(self): + lock = self.lock() + await lock.get() + + await lock.release() + + self.assertEqual(Path(lock.path).read_bytes(), b"") + + +class TestDatabaseLock(LockContract, IsolatedAsyncioTestCase): + async def make_store(self): + self.database = InMemoryDatabase() + self.addAsyncCleanup(self.database.engine.dispose) + migration = load_migration() + with mock.patch.object(migration, "config", return_value=None): + await migration.CreateCachesTable("sqlite", self.database.get_schema_builder()).up() + return DatabaseStore(self.database.connection) + + async def test_lock_row_does_not_collide_with_cache_key(self): + await self.store.put("reports", "cached", None) + + self.assertIs(await self.lock().get(), True) + self.assertEqual(await self.store.get("reports"), "cached") + + +class TestRepositoryLock(IsolatedAsyncioTestCase): + async def test_prefixes_lock_name(self): + store = ArrayStore() + + await Repository(store, "app").lock("reports", 10).get() + + self.assertIn("app:reports", store.locks) + + async def test_store_without_locks_raises(self): + class PlainStore: + pass + + with self.assertRaises(TypeError): + Repository(PlainStore()).lock("reports") # pyright: ignore[reportArgumentType] From 9bb0f71615fc3786e7ca2dcd69099e596c5675c5 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Wed, 21 Oct 2026 12:35:19 -0700 Subject: [PATCH 2/6] Make lock release consistent for expired locks and cover concurrency and async with errors --- docs/adr/031-cache-locks.md | 6 ++--- .../fastapi_startkit/cache/database_store.py | 8 ++++++- fastapi_startkit/tests/cache/test_lock.py | 22 +++++++++++++++++++ 3 files changed, 32 insertions(+), 4 deletions(-) diff --git a/docs/adr/031-cache-locks.md b/docs/adr/031-cache-locks.md index 7cf216c3..9256a1c6 100644 --- a/docs/adr/031-cache-locks.md +++ b/docs/adr/031-cache-locks.md @@ -20,12 +20,12 @@ Applications need to make sure only one worker runs a piece of work at a time (r - `await lock.get()` tries once and returns `bool`; `await lock.get(callback)` runs the sync or async callback only when acquired, releases afterwards (also on error) and returns the callback result, or `False` when not acquired. - `await lock.block(seconds, callback=None)` retries every 250 ms (`lock.sleep`) until acquired or `seconds` passed, then raises `LockTimeoutError`; with a callback it runs and releases like `get`. - `async with Cache.lock("name", 10):` acquires once or raises `LockTimeoutError`, and releases on exit. Use `block` to wait. - - `release()` deletes the lock only when this owner holds it and returns `bool`; `force_release()` deletes it regardless of owner. + - `release()` deletes the lock only when this owner holds it and it has not expired, and returns `bool` (an owner's own expired lock returns `False` on every store); `force_release()` deletes it regardless of owner. - Locks are not re-entrant: acquiring a lock the same owner already holds fails, on every store. - Store locks: - `ArrayStore`: a `locks` dict of `name -> (owner, expires_at)` on the store. Check and set run without an `await`, so they are atomic within the event loop. - `FileStore`: one file per lock under `/locks/`, holding JSON `[owner, expires_at]`. Each operation opens the file, takes an exclusive `fcntl.flock`, reads, decides, rewrites or truncates it, and closes it, so acquire and owner-checked release are atomic across processes on one host. Files are truncated, never unlinked, because unlinking a file another process has open would let both lock different inodes. Lock files live outside the two-hex shard directories, so `flush` does not touch them. `fcntl` is imported on use, so the file cache still imports on Windows; locks on the file store are POSIX-only. - - `DatabaseStore`: a row in the cache table with key `lock:`, the pickled owner as value and the usual expiration column. Acquire inserts; on a primary-key conflict it updates the row only where `expiration <= now`, and the lock is acquired when one row changed. Release deletes where key and value match the owner. Reusing the cache table needs no new migration; `flush` therefore also clears database locks. The PostgreSQL transaction caveat of ADR 029 applies to the insert conflict. + - `DatabaseStore`: a row in the cache table with key `lock:`, the pickled owner as value and the usual expiration column. Acquire inserts; on a primary-key conflict it updates the row only where `expiration <= now`, and the lock is acquired when one row changed. Release deletes where key and value match the owner and the row has not expired. Reusing the cache table needs no new migration; `flush` therefore also clears database locks. The PostgreSQL transaction caveat of ADR 029 applies to the insert conflict. - Redis follows in a separate PR on top of the Redis store (`SET NX EX` and an owner-checked delete script). - `Repository.lock(name, seconds=0, owner=None)` prefixes the name like a cache key and returns the store's lock; a store without locks raises `TypeError`. The facade reaches it through the manager; the stub adds `lock`. @@ -53,4 +53,4 @@ await Cache.lock("reports", 10, owner).release() ## Validation -`tests/cache/test_lock.py` (unittest) runs one shared case class against the array, file and database (SQLite) stores: acquire once, second owner fails, same owner fails, expired lock can be taken, `seconds=0` never expires, release only by owner, force release, restore by owner, `get` with sync/async callback releases also on error, `block` waits for a release and times out with `LockTimeoutError`, `async with` acquires/releases and raises when held, the repository prefixes lock names, and a store without locks raises `TypeError`. The file store also checks that `flush` keeps locks and that a corrupt lock file counts as free. +`tests/cache/test_lock.py` (unittest) runs one shared case class against the array, file and database (SQLite) stores: acquire once, exactly one winner among concurrent `asyncio.gather` acquires, second owner fails, same owner fails, expired lock can be taken, `seconds=0` never expires, release only by owner, releasing an own expired lock returns `False`, force release, restore by owner, `get` with sync/async callback releases also on error, `block` waits for a release and times out with `LockTimeoutError`, `async with` acquires/releases, releases when the body raises, and raises when held, the repository prefixes lock names, and a store without locks raises `TypeError`. The file store also checks that `flush` keeps locks and that a corrupt lock file counts as free. diff --git a/fastapi_startkit/src/fastapi_startkit/cache/database_store.py b/fastapi_startkit/src/fastapi_startkit/cache/database_store.py index a10e118e..d182c5f2 100644 --- a/fastapi_startkit/src/fastapi_startkit/cache/database_store.py +++ b/fastapi_startkit/src/fastapi_startkit/cache/database_store.py @@ -82,7 +82,13 @@ async def acquire(self) -> bool: return True async def release(self) -> bool: - return bool(await self.store.query().where("key", self.key).where("value", encode(self.owner)).delete()) + held = ( + self.store.query() + .where("key", self.key) + .where("value", encode(self.owner)) + .where(lambda query: query.where_null("expiration").or_where("expiration", ">", time.time())) + ) + return bool(await held.delete()) async def force_release(self) -> None: await self.store.query().where("key", self.key).delete() diff --git a/fastapi_startkit/tests/cache/test_lock.py b/fastapi_startkit/tests/cache/test_lock.py index 6aad7459..6bb77bb0 100644 --- a/fastapi_startkit/tests/cache/test_lock.py +++ b/fastapi_startkit/tests/cache/test_lock.py @@ -56,6 +56,21 @@ async def test_release_only_by_owner(self): self.assertIs(await lock.release(), False) self.assertIs(await self.lock().get(), True) + async def test_release_of_own_expired_lock_returns_false(self): + lock = self.lock(seconds=10) + await lock.get() + + self.time.return_value += 11 + + self.assertIs(await lock.release(), False) + self.assertIs(await self.lock().get(), True) + + async def test_concurrent_acquires_have_exactly_one_winner(self): + results = await asyncio.gather(*(self.lock().get() for _ in range(10))) + + self.assertEqual(results.count(True), 1) + self.assertEqual(results.count(False), 9) + async def test_force_release_ignores_owner(self): await self.lock().get() @@ -129,6 +144,13 @@ async def test_async_with_acquires_and_releases(self): self.assertIs(await self.lock().get(), True) + async def test_async_with_releases_when_body_raises(self): + with self.assertRaises(RuntimeError): + async with self.lock(): + raise RuntimeError("boom") + + self.assertIs(await self.lock().get(), True) + async def test_async_with_raises_when_held(self): await self.lock().get() From ea1cc560ec350080ca0c9087948f7e0f094d8d76 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Wed, 21 Oct 2026 12:33:16 -0700 Subject: [PATCH 3/6] Add Cache.tags() namespaced tagged cache with tagged flush --- docs/adr/034-cache-tags.md | 41 ++++++ docs/adr/AGENTS.md | 1 + .../src/fastapi_startkit/cache/__init__.py | 3 + .../src/fastapi_startkit/cache/repository.py | 9 +- .../fastapi_startkit/cache/tagged_store.py | 57 ++++++++ .../src/fastapi_startkit/facades/Cache.pyi | 3 +- .../tests/cache/test_tagged_store.py | 128 ++++++++++++++++++ 7 files changed, 240 insertions(+), 2 deletions(-) create mode 100644 docs/adr/034-cache-tags.md create mode 100644 fastapi_startkit/src/fastapi_startkit/cache/tagged_store.py create mode 100644 fastapi_startkit/tests/cache/test_tagged_store.py diff --git a/docs/adr/034-cache-tags.md b/docs/adr/034-cache-tags.md new file mode 100644 index 00000000..6d6a37d2 --- /dev/null +++ b/docs/adr/034-cache-tags.md @@ -0,0 +1,41 @@ +--- +title: Tagged cache entries with Cache.tags() +date: 2026-10-21 +--- + +## Problem + +Applications cache many related entries (every page of a user's posts, every report for a team) and need to invalidate the whole group at once without knowing every key. Laravel offers `Cache::tags([...])`: entries written through a tagged cache can be read only with the same tags, and `Cache::tags([...])->flush()` invalidates every entry that carries any of those tags. The cache module has no grouping, and `flush()` clears the whole store. + +## Alternatives + +- **Store a key list per tag and delete each key on flush.** Every store needs a list or set type and an atomic append, and the file and database stores have neither. Lists also grow without bound for keys that expire. Rejected. +- **Scan keys by pattern on flush.** Not possible for the file store (keys are hashed) and slow for the database. Rejected. +- **Namespaced keys (Laravel's original tagged cache).** Each tag has a random version id stored in the cache. A tagged entry's key is prefixed with a hash of its tags' ids. Flushing a tag writes a new id, so old entries can no longer be addressed. Works on every store with plain `get`/`put`. Chosen. + +## Decision + +- `cache/tagged_store.py` adds `TagSet(store, names, key)` and `TaggedStore(store, tags)`. + - `TagSet` keeps the names deduplicated and sorted, so `tags(["a", "b"])` and `tags(["b", "a"])` address the same entries (Laravel is order-sensitive; sorting removes that trap). The id of tag `name` lives in the wrapped store under the repository key `tag::key` (`:tag::key` with a prefix), stored forever, and is created as a random hex token on first use. `namespace()` returns the SHA-1 of the ids joined with `|`; `reset()` writes a new id for every tag. + - `TaggedStore` implements the store contract over the wrapped store: `get`, `put` and `forget` use the key `:`, and `flush` resets the tags instead of flushing the store. +- `Repository.tags(names)` accepts a string or a sequence of strings and returns a `Repository` over `TaggedStore` with the same prefix, so every repository method (`get`, `put`, `remember`, the decorator, ...) works tagged. It wraps whatever store the repository has, so `Cache.memo().tags(...)` and `Cache.store("file").tags(...)` work too. An empty tag list raises `ValueError`. A tagged repository has no `lock`. +- Flushed entries are not deleted; they stay in the store until they expire or the store is flushed. Use a TTL on tagged entries in long-lived stores. If two workers create a tag's first id at the same time, the last write wins, and entries the other worker wrote under its id become misses. That only costs a recompute. +- Reading a tagged entry costs one extra store read per tag. +- The facade reaches `tags` through the manager; the stub adds `tags`. + +## Usage + +```python +from fastapi_startkit.facades import Cache + +await Cache.tags(["people", "artists"]).put("john", john, 600) +john = await Cache.tags(["people", "artists"]).get("john") + +posts = await Cache.tags("posts").remember(f"user:{user_id}", 600, load_posts) + +await Cache.tags("people").flush() +``` + +## Validation + +`tests/cache/test_tagged_store.py` (unittest, array and file stores): tagged put/get round-trips and untagged or differently tagged reads miss; tag order and duplicates do not matter; flushing one tag invalidates every entry carrying it and keeps entries without it, and keeps untagged entries; tag ids persist in the store and are prefixed; forget removes only the tagged key; `remember` and the decorator work tagged; a string tag is accepted; an empty list raises; tags work on a memo repository. diff --git a/docs/adr/AGENTS.md b/docs/adr/AGENTS.md index a9b24b8e..f69299bc 100644 --- a/docs/adr/AGENTS.md +++ b/docs/adr/AGENTS.md @@ -34,3 +34,4 @@ Read this index first, then open the records relevant to the change. Before impl | [029](029-cache-module.md) | 2026-10-21 | Cache module with callable remember and pluggable stores | New `fastapi_startkit.cache`: async `Cache.get/put/has/forget/flush`, `get` with a callable default, and `remember(key, ttl, factory)` that runs a sync or async factory only on a miss; `remember(key, ttl)` without a factory is a decorator keyed by `key` plus a hash of the call arguments. Stores implement an async contract with a `MISSING` sentinel (cached `None` is a hit): array, file (pickled per-key files) and database (masoniteorm table). TTL in seconds or `timedelta`, `None` forever, <=0 forgets. `CacheConfig` (`CACHE_CONNECTION`, `CACHE_PREFIX`, `stores`) via `resolve_config`; `CacheProvider` binds `cache`; facade `remember` resolves lazily. A `with` form is impossible in Python. Shipped in slices into `feature/caches`. | | [030](030-cache-memo.md) | 2026-10-21 | Request-scoped cache memoization with Cache.memo() | `Cache.memo(name=None)` returns a `Repository` over `MemoStore`, which wraps the named store and memoizes reads (misses too) for the current asyncio task; `put`/`forget` write through and drop the key, `flush` clears memo and store. State is a `ContextVar` holding `(weakref to owning task, memo)`, and a task that finds another task's memo starts a fresh one, so requests, jobs and child tasks never share memo state. Memoized values ignore TTL for the rest of the task. | | [031](031-cache-locks.md) | 2026-10-21 | Atomic cache locks with Cache.lock() | `Cache.lock(name, seconds=0, owner=None)` returns a store-specific `Lock` (abstract base in `cache/lock.py`): `get(callback=None)` tries once, `block(seconds, callback=None)` polls every 250 ms then raises `LockTimeoutError`, `async with` acquires once or raises, `release()` is owner-checked, `force_release()` is not; callbacks release on error; passing `owner` restores a lock elsewhere; not re-entrant; `seconds<=0` never expires. Array: dict on the store, atomic in the event loop. File: `/locks/` JSON `[owner, expires_at]` under `fcntl.flock`, truncated never unlinked, kept by `flush`, POSIX-only. Database: row `lock:` in the cache table, insert then update-if-expired on conflict, release deletes by owner; no new migration, `flush` clears it. Redis lock follows on the Redis store. | +| [034](034-cache-tags.md) | 2026-10-21 | Tagged cache entries with Cache.tags() | `Cache.tags(name or names)` returns a `Repository` over `TaggedStore`, using Laravel's namespaced-key design. Each tag has a random id stored forever at `:tag::key`. A tagged entry's key is `sha1(ids joined by "|"):`, and `flush()` writes new ids, so every entry carrying any flushed tag becomes unreachable on every store with plain get/put. Tags are deduplicated and sorted, so order does not matter. Flushed entries stay until they expire or the store is flushed. An empty list raises `ValueError`. It composes with `memo()` and named stores. | diff --git a/fastapi_startkit/src/fastapi_startkit/cache/__init__.py b/fastapi_startkit/src/fastapi_startkit/cache/__init__.py index c55bc155..37873506 100644 --- a/fastapi_startkit/src/fastapi_startkit/cache/__init__.py +++ b/fastapi_startkit/src/fastapi_startkit/cache/__init__.py @@ -7,6 +7,7 @@ from .provider import CacheProvider from .repository import Repository from .store import MISSING, Store +from .tagged_store import TaggedStore, TagSet __all__ = [ "MISSING", @@ -23,4 +24,6 @@ "MemoStore", "Repository", "Store", + "TagSet", + "TaggedStore", ] diff --git a/fastapi_startkit/src/fastapi_startkit/cache/repository.py b/fastapi_startkit/src/fastapi_startkit/cache/repository.py index 115414a6..5cf9b623 100644 --- a/fastapi_startkit/src/fastapi_startkit/cache/repository.py +++ b/fastapi_startkit/src/fastapi_startkit/cache/repository.py @@ -2,11 +2,12 @@ import hashlib import inspect import pickle -from collections.abc import Awaitable, Callable +from collections.abc import Awaitable, Callable, Sequence from datetime import timedelta from typing import TYPE_CHECKING, Any, ParamSpec, Protocol, overload from .store import MISSING, Store +from .tagged_store import TaggedStore, TagSet if TYPE_CHECKING: from .lock import Lock @@ -115,3 +116,9 @@ def lock(self, name: str, seconds: float = 0, owner: str | None = None) -> "Lock if factory is None: raise TypeError(f"Cache store [{type(self.store).__name__}] does not support locks.") return factory(self.key(name), seconds, owner) + + def tags(self, names: str | Sequence[str]) -> "Repository": + tags = TagSet(self.store, [names] if isinstance(names, str) else names, self.key) + if not tags.names: + raise ValueError("Cache.tags requires at least one tag.") + return Repository(TaggedStore(self.store, tags), self.prefix) diff --git a/fastapi_startkit/src/fastapi_startkit/cache/tagged_store.py b/fastapi_startkit/src/fastapi_startkit/cache/tagged_store.py new file mode 100644 index 00000000..d1d5fa61 --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/cache/tagged_store.py @@ -0,0 +1,57 @@ +import hashlib +import secrets +from collections.abc import Callable, Iterable +from typing import Any + +from .store import MISSING, Store + + +class TagSet: + def __init__(self, store: Store, names: Iterable[str], key: Callable[[str], str]): + self.store = store + self.names = sorted(set(names)) + self.key = key + + def tag_key(self, name: str) -> str: + return self.key(f"tag:{name}:key") + + async def tag_id(self, name: str) -> str: + identifier = await self.store.get(self.tag_key(name)) + if identifier is MISSING: + return await self.reset_tag(name) + return identifier + + async def reset_tag(self, name: str) -> str: + identifier = secrets.token_hex(12) + await self.store.put(self.tag_key(name), identifier, None) + return identifier + + async def namespace(self) -> str: + identifiers = [await self.tag_id(name) for name in self.names] + return hashlib.sha1("|".join(identifiers).encode()).hexdigest() + + async def reset(self) -> None: + for name in self.names: + await self.reset_tag(name) + + +class TaggedStore: + def __init__(self, store: Store, tags: TagSet): + self.store = store + self.tags = tags + + async def tagged(self, key: str) -> str: + return f"{await self.tags.namespace()}:{key}" + + async def get(self, key: str) -> Any: + return await self.store.get(await self.tagged(key)) + + async def put(self, key: str, value: Any, seconds: float | None) -> bool: + return await self.store.put(await self.tagged(key), value, seconds) + + async def forget(self, key: str) -> bool: + return await self.store.forget(await self.tagged(key)) + + async def flush(self) -> bool: + await self.tags.reset() + return True diff --git a/fastapi_startkit/src/fastapi_startkit/facades/Cache.pyi b/fastapi_startkit/src/fastapi_startkit/facades/Cache.pyi index 6264dcb3..aad7cc9e 100644 --- a/fastapi_startkit/src/fastapi_startkit/facades/Cache.pyi +++ b/fastapi_startkit/src/fastapi_startkit/facades/Cache.pyi @@ -1,4 +1,4 @@ -from collections.abc import Awaitable, Callable +from collections.abc import Awaitable, Callable, Sequence from typing import Any, overload from fastapi_startkit.cache import CacheManager, Lock, Repository @@ -8,6 +8,7 @@ from fastapi_startkit.cache.repository import Decorator, Ttl class CacheFacade: def store(self, name: str | None = None) -> Repository: ... def memo(self, name: str | None = None) -> Repository: ... + def tags(self, names: str | Sequence[str]) -> Repository: ... def extend(self, driver: str, factory: StoreFactory) -> CacheManager: ... async def get(self, key: str, default: Any = None) -> Any: ... async def has(self, key: str) -> bool: ... diff --git a/fastapi_startkit/tests/cache/test_tagged_store.py b/fastapi_startkit/tests/cache/test_tagged_store.py new file mode 100644 index 00000000..27187148 --- /dev/null +++ b/fastapi_startkit/tests/cache/test_tagged_store.py @@ -0,0 +1,128 @@ +import tempfile +from unittest import IsolatedAsyncioTestCase + +from fastapi_startkit.cache import ArrayStore, CacheManager, FileStore, MemoStore, Repository, TaggedStore + + +class TaggedContract: + def make_store(self): + raise NotImplementedError + + def setUp(self): + self.store = self.make_store() + self.cache = Repository(self.store, "app") + + async def test_tagged_put_and_get(self): + self.assertIs(await self.cache.tags(["people", "artists"]).put("john", "doe", 600), True) + + self.assertEqual(await self.cache.tags(["people", "artists"]).get("john"), "doe") + + async def test_untagged_and_other_tags_miss(self): + await self.cache.tags(["people", "artists"]).put("john", "doe") + + self.assertIsNone(await self.cache.get("john")) + self.assertIsNone(await self.cache.tags("people").get("john")) + self.assertIsNone(await self.cache.tags(["people", "authors"]).get("john")) + + async def test_tag_order_and_duplicates_do_not_matter(self): + await self.cache.tags(["people", "artists"]).put("john", "doe") + + self.assertEqual(await self.cache.tags(["artists", "people", "artists"]).get("john"), "doe") + + async def test_flush_invalidates_entries_carrying_any_flushed_tag(self): + await self.cache.tags(["people", "artists"]).put("john", "doe") + await self.cache.tags(["people", "authors"]).put("anne", "rice") + await self.cache.tags("authors").put("mark", "twain") + await self.cache.put("plain", "value") + + self.assertIs(await self.cache.tags("people").flush(), True) + + self.assertIsNone(await self.cache.tags(["people", "artists"]).get("john")) + self.assertIsNone(await self.cache.tags(["people", "authors"]).get("anne")) + self.assertEqual(await self.cache.tags("authors").get("mark"), "twain") + self.assertEqual(await self.cache.get("plain"), "value") + + async def test_tag_ids_persist_in_the_store_with_prefix(self): + await self.cache.tags("people").put("john", "doe") + identifier = await self.cache.get("tag:people:key") + + self.assertIsInstance(identifier, str) + self.assertEqual(await Repository(self.store, "app").tags("people").get("john"), "doe") + + await self.cache.tags("people").flush() + + self.assertNotEqual(await self.cache.get("tag:people:key"), identifier) + + async def test_forget_removes_only_the_tagged_key(self): + await self.cache.tags("people").put("john", "doe") + await self.cache.tags("people").put("jane", "roe") + await self.cache.put("john", "plain") + + self.assertIs(await self.cache.tags("people").forget("john"), True) + + self.assertIsNone(await self.cache.tags("people").get("john")) + self.assertEqual(await self.cache.tags("people").get("jane"), "roe") + self.assertEqual(await self.cache.get("john"), "plain") + + async def test_remember_is_tagged(self): + calls = [] + + def load(): + calls.append(1) + return ["post"] + + self.assertEqual(await self.cache.tags("posts").remember("user:1", 600, load), ["post"]) + self.assertEqual(await self.cache.tags("posts").remember("user:1", 600, load), ["post"]) + self.assertEqual(len(calls), 1) + + await self.cache.tags("posts").flush() + + await self.cache.tags("posts").remember("user:1", 600, load) + self.assertEqual(len(calls), 2) + + async def test_decorator_is_tagged(self): + calls = [] + + @self.cache.tags("users").remember("user", 600) + async def find(user_id): + calls.append(user_id) + return {"id": user_id} + + self.assertEqual(await find(1), {"id": 1}) + self.assertEqual(await find(1), {"id": 1}) + self.assertEqual(calls, [1]) + + +class TestArrayTags(TaggedContract, IsolatedAsyncioTestCase): + def make_store(self): + return ArrayStore() + + +class TestFileTags(TaggedContract, IsolatedAsyncioTestCase): + def make_store(self): + return FileStore(self.enterContext(tempfile.TemporaryDirectory())) + + +class TestRepositoryTags(IsolatedAsyncioTestCase): + async def test_tags_wraps_store_and_keeps_prefix(self): + store = ArrayStore() + + tagged = Repository(store, "app").tags(["b", "a", "b"]) + + self.assertIsInstance(tagged.store, TaggedStore) + self.assertIs(tagged.store.store, store) + self.assertEqual(tagged.store.tags.names, ["a", "b"]) + self.assertEqual(tagged.prefix, "app") + + async def test_empty_tags_raise(self): + with self.assertRaises(ValueError): + Repository(ArrayStore()).tags([]) + + async def test_tags_work_on_memo_repository(self): + manager = CacheManager({"default": "array", "stores": {"array": {"driver": "array"}}}) + + await manager.memo().tags("people").put("john", "doe") + + self.assertIsInstance(manager.memo().store, MemoStore) + self.assertEqual(await manager.tags("people").get("john"), "doe") + self.assertEqual(await manager.memo().tags("people").get("john"), "doe") From 251133d1bfabdf65a9e04422b67c4ed879c69c14 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Wed, 21 Oct 2026 12:36:23 -0700 Subject: [PATCH 4/6] Add Redis cache lock with SET NX PX and owner-checked release script --- docs/adr/031-cache-locks.md | 4 +-- docs/adr/AGENTS.md | 2 +- .../src/fastapi_startkit/cache/redis_store.py | 27 +++++++++++++++ fastapi_startkit/tests/cache/test_lock.py | 34 +++++++++++++++++++ 4 files changed, 64 insertions(+), 3 deletions(-) diff --git a/docs/adr/031-cache-locks.md b/docs/adr/031-cache-locks.md index 9256a1c6..fe424354 100644 --- a/docs/adr/031-cache-locks.md +++ b/docs/adr/031-cache-locks.md @@ -26,7 +26,7 @@ Applications need to make sure only one worker runs a piece of work at a time (r - `ArrayStore`: a `locks` dict of `name -> (owner, expires_at)` on the store. Check and set run without an `await`, so they are atomic within the event loop. - `FileStore`: one file per lock under `/locks/`, holding JSON `[owner, expires_at]`. Each operation opens the file, takes an exclusive `fcntl.flock`, reads, decides, rewrites or truncates it, and closes it, so acquire and owner-checked release are atomic across processes on one host. Files are truncated, never unlinked, because unlinking a file another process has open would let both lock different inodes. Lock files live outside the two-hex shard directories, so `flush` does not touch them. `fcntl` is imported on use, so the file cache still imports on Windows; locks on the file store are POSIX-only. - `DatabaseStore`: a row in the cache table with key `lock:`, the pickled owner as value and the usual expiration column. Acquire inserts; on a primary-key conflict it updates the row only where `expiration <= now`, and the lock is acquired when one row changed. Release deletes where key and value match the owner and the row has not expired. Reusing the cache table needs no new migration; `flush` therefore also clears database locks. The PostgreSQL transaction caveat of ADR 029 applies to the insert conflict. - - Redis follows in a separate PR on top of the Redis store (`SET NX EX` and an owner-checked delete script). + - `RedisStore`: key `lock:` (the Redis connection prefix applies) holding the raw owner. Acquire is `SET NX PX` (no `PX` when the lock never expires). Release is a Lua script that deletes the key only while it holds this owner, and force release is `DEL`. Redis expires the key itself, so releasing your own expired lock returns `False`. The key is outside the store's `:` namespace, so `flush` keeps Redis locks. The key also never collides with a cache entry, whose corrupt-value handling would otherwise delete the lock. - `Repository.lock(name, seconds=0, owner=None)` prefixes the name like a cache key and returns the store's lock; a store without locks raises `TypeError`. The facade reaches it through the manager; the stub adds `lock`. ## Usage @@ -53,4 +53,4 @@ await Cache.lock("reports", 10, owner).release() ## Validation -`tests/cache/test_lock.py` (unittest) runs one shared case class against the array, file and database (SQLite) stores: acquire once, exactly one winner among concurrent `asyncio.gather` acquires, second owner fails, same owner fails, expired lock can be taken, `seconds=0` never expires, release only by owner, releasing an own expired lock returns `False`, force release, restore by owner, `get` with sync/async callback releases also on error, `block` waits for a release and times out with `LockTimeoutError`, `async with` acquires/releases, releases when the body raises, and raises when held, the repository prefixes lock names, and a store without locks raises `TypeError`. The file store also checks that `flush` keeps locks and that a corrupt lock file counts as free. +`tests/cache/test_lock.py` (unittest) runs one shared case class against the array, file, database (SQLite) and Redis (fakeredis with Lua) stores: acquire once, exactly one winner among concurrent `asyncio.gather` acquires, second owner fails, same owner fails, expired lock can be taken, `seconds=0` never expires, release only by owner, releasing an own expired lock returns `False`, force release, restore by owner, `get` with sync/async callback releases also on error, `block` waits for a release and times out with `LockTimeoutError`, `async with` acquires/releases, releases when the body raises, and raises when held, the repository prefixes lock names, and a store without locks raises `TypeError`. The file store also checks that `flush` keeps locks and that a corrupt lock file counts as free. The Redis store checks the prefixed key and its TTL, that the lock does not collide with a cache key, and that `flush` keeps locks. diff --git a/docs/adr/AGENTS.md b/docs/adr/AGENTS.md index 46998174..65393e52 100644 --- a/docs/adr/AGENTS.md +++ b/docs/adr/AGENTS.md @@ -33,4 +33,4 @@ Read this index first, then open the records relevant to the change. Before impl | [025](025-model-lifecycle-events.md) | 2026-10-20 | Model lifecycle events, automatic timestamps and datetime casts | `save()` fires `saving → creating/updating → query → created/updated → saved`. Observer changes are persisted, and `updating` fires only when the model is dirty. A new `Model.delete()` fires `deleting → DELETE → deleted`. `observe_events` is async and awaits async handlers. Mass query-builder writes fire no events. Laravel-style auto timestamps: `created_at`/`updated_at` are filled on create unless supplied, `updated_at` moves on update, and `__timestamps__ = False` disables both. `DateCast.set` returns a UTC datetime; `Connection.prepare_binding` passes it unchanged to Postgres/MySQL, and SQLite converts it to a naive UTC string. Postgres registers an asyncpg `timestamp` codec that stores aware values as naive UTC; `timestamptz` keeps asyncpg's codec. `Annotated[T, Cast]` (a `BaseCast` subclass or instance) overrides a column's cast, and `Attribute.__casts__` is removed. | | [029](029-cache-module.md) | 2026-10-21 | Cache module with callable remember and pluggable stores | New `fastapi_startkit.cache`: async `Cache.get/put/has/forget/flush`, `get` with a callable default, and `remember(key, ttl, factory)` that runs a sync or async factory only on a miss; `remember(key, ttl)` without a factory is a decorator keyed by `key` plus a hash of the call arguments. Stores implement an async contract with a `MISSING` sentinel (cached `None` is a hit): array, file (pickled per-key files), database (masoniteorm table) and redis (named connection from the existing `RedisManager`, `SET PX` TTL, flush by `SCAN` of prefixed keys, never `FLUSHDB`). TTL in seconds or `timedelta`, `None` forever, <=0 forgets. `CacheConfig` (`CACHE_CONNECTION`, `CACHE_PREFIX`, `stores`) via `resolve_config`; `CacheProvider` binds `cache`; facade `remember` resolves lazily. A `with` form is impossible in Python. Shipped in slices into `feature/caches`. | | [030](030-cache-memo.md) | 2026-10-21 | Request-scoped cache memoization with Cache.memo() | `Cache.memo(name=None)` returns a `Repository` over `MemoStore`, which wraps the named store and memoizes reads (misses too) for the current asyncio task; `put`/`forget` write through and drop the key, `flush` clears memo and store. State is a `ContextVar` holding `(weakref to owning task, memo)`, and a task that finds another task's memo starts a fresh one, so requests, jobs and child tasks never share memo state. Memoized values ignore TTL for the rest of the task. | -| [031](031-cache-locks.md) | 2026-10-21 | Atomic cache locks with Cache.lock() | `Cache.lock(name, seconds=0, owner=None)` returns a store-specific `Lock` (abstract base in `cache/lock.py`): `get(callback=None)` tries once, `block(seconds, callback=None)` polls every 250 ms then raises `LockTimeoutError`, `async with` acquires once or raises, `release()` is owner-checked, `force_release()` is not; callbacks release on error; passing `owner` restores a lock elsewhere; not re-entrant; `seconds<=0` never expires. Array: dict on the store, atomic in the event loop. File: `/locks/` JSON `[owner, expires_at]` under `fcntl.flock`, truncated never unlinked, kept by `flush`, POSIX-only. Database: row `lock:` in the cache table, insert then update-if-expired on conflict, release deletes by owner; no new migration, `flush` clears it. Redis lock follows on the Redis store. | +| [031](031-cache-locks.md) | 2026-10-21 | Atomic cache locks with Cache.lock() | `Cache.lock(name, seconds=0, owner=None)` returns a store-specific `Lock` (abstract base in `cache/lock.py`): `get(callback=None)` tries once, `block(seconds, callback=None)` polls every 250 ms then raises `LockTimeoutError`, `async with` acquires once or raises, `release()` is owner-checked, `force_release()` is not; callbacks release on error; passing `owner` restores a lock elsewhere; not re-entrant; `seconds<=0` never expires. Array: dict on the store, atomic in the event loop. File: `/locks/` JSON `[owner, expires_at]` under `fcntl.flock`, truncated never unlinked, kept by `flush`, POSIX-only. Database: row `lock:` in the cache table, insert then update-if-expired on conflict, release deletes by owner; no new migration, `flush` clears it. Redis: `lock:` via `SET NX PX` and an owner-checked Lua delete, kept by `flush`. | diff --git a/fastapi_startkit/src/fastapi_startkit/cache/redis_store.py b/fastapi_startkit/src/fastapi_startkit/cache/redis_store.py index 439ec54b..f923248f 100644 --- a/fastapi_startkit/src/fastapi_startkit/cache/redis_store.py +++ b/fastapi_startkit/src/fastapi_startkit/cache/redis_store.py @@ -7,9 +7,16 @@ from fastapi_startkit.redis import Connection +from .lock import Lock from .store import MISSING GLOB = re.compile(r"([*?\[\]\\])") +RELEASE = """ +if redis.call("get", KEYS[1]) == ARGV[1] then + return redis.call("del", KEYS[1]) +end +return 0 +""" class RedisStore: @@ -52,6 +59,26 @@ async def flush(self) -> bool: await connection.unlink(*keys) return True + def lock(self, name: str, seconds: float = 0, owner: str | None = None) -> "RedisLock": + return RedisLock(self, name, seconds, owner) + + +class RedisLock(Lock): + def __init__(self, store: RedisStore, name: str, seconds: float = 0, owner: str | None = None): + super().__init__(name, seconds, owner) + self.store = store + self.key = f"lock:{name}" + + async def acquire(self) -> bool: + milliseconds = None if self.ttl is None else max(1, math.ceil(self.ttl * 1000)) + return bool(await self.store.connection().set(self.key, self.owner, nx=True, px=milliseconds)) + + async def release(self) -> bool: + return bool(await self.store.connection().eval(RELEASE, 1, self.key, self.owner)) + + async def force_release(self) -> None: + await self.store.connection().delete(self.key) + def strip(key: Any, prefix: str) -> Any: marker = prefix.encode() if isinstance(key, bytes) else prefix diff --git a/fastapi_startkit/tests/cache/test_lock.py b/fastapi_startkit/tests/cache/test_lock.py index 6bb77bb0..1503b8a6 100644 --- a/fastapi_startkit/tests/cache/test_lock.py +++ b/fastapi_startkit/tests/cache/test_lock.py @@ -3,8 +3,12 @@ from pathlib import Path from unittest import IsolatedAsyncioTestCase, mock +import fakeredis + from fastapi_startkit.cache import ArrayStore, FileStore, LockTimeoutError, Repository from fastapi_startkit.cache.database_store import DatabaseStore +from fastapi_startkit.cache.redis_store import RedisStore +from fastapi_startkit.redis import RedisManager from test_database_store import InMemoryDatabase, load_migration @@ -208,6 +212,36 @@ async def test_lock_row_does_not_collide_with_cache_key(self): self.assertEqual(await self.store.get("reports"), "cached") +class TestRedisLock(LockContract, IsolatedAsyncioTestCase): + async def make_store(self): + self.server = fakeredis.FakeServer() + manager = RedisManager( + {"client": "fake", "options": {"prefix": "app_"}, "connections": {"cache": {"database": 1}}} + ).extend("fake", lambda parameters: fakeredis.FakeAsyncRedis(server=self.server, **parameters)) + self.raw = fakeredis.FakeAsyncRedis(server=self.server, db=1, decode_responses=True) + return RedisStore(lambda: manager.connection("cache"), "shop") + + async def test_lock_key_uses_connection_prefix_and_expiry(self): + lock = self.lock(seconds=10) + await lock.get() + + self.assertEqual(await self.raw.get("app_lock:reports"), lock.owner) + self.assertGreater(await self.raw.pttl("app_lock:reports"), 0) + + async def test_lock_does_not_collide_with_cache_key(self): + await self.store.put("reports", "cached", None) + + self.assertIs(await self.lock().get(), True) + self.assertEqual(await self.store.get("reports"), "cached") + + async def test_flush_keeps_locks(self): + await self.lock().get() + + await self.store.flush() + + self.assertIs(await self.lock().get(), False) + + class TestRepositoryLock(IsolatedAsyncioTestCase): async def test_prefixes_lock_name(self): store = ArrayStore() From 62f5438b1ba14e48fbcd8ad988aa09c9e64f6d00 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Wed, 21 Oct 2026 12:37:26 -0700 Subject: [PATCH 5/6] Put cache prefix first in tagged keys and cover database store and lock on tagged cache --- docs/adr/034-cache-tags.md | 7 +-- docs/adr/AGENTS.md | 2 +- .../src/fastapi_startkit/cache/repository.py | 2 +- .../fastapi_startkit/cache/tagged_store.py | 2 +- .../tests/cache/test_tagged_store.py | 48 ++++++++++++++++--- 5 files changed, 48 insertions(+), 13 deletions(-) diff --git a/docs/adr/034-cache-tags.md b/docs/adr/034-cache-tags.md index 6d6a37d2..48f6a309 100644 --- a/docs/adr/034-cache-tags.md +++ b/docs/adr/034-cache-tags.md @@ -17,8 +17,9 @@ Applications cache many related entries (every page of a user's posts, every rep - `cache/tagged_store.py` adds `TagSet(store, names, key)` and `TaggedStore(store, tags)`. - `TagSet` keeps the names deduplicated and sorted, so `tags(["a", "b"])` and `tags(["b", "a"])` address the same entries (Laravel is order-sensitive; sorting removes that trap). The id of tag `name` lives in the wrapped store under the repository key `tag::key` (`:tag::key` with a prefix), stored forever, and is created as a random hex token on first use. `namespace()` returns the SHA-1 of the ids joined with `|`; `reset()` writes a new id for every tag. - - `TaggedStore` implements the store contract over the wrapped store: `get`, `put` and `forget` use the key `:`, and `flush` resets the tags instead of flushing the store. -- `Repository.tags(names)` accepts a string or a sequence of strings and returns a `Repository` over `TaggedStore` with the same prefix, so every repository method (`get`, `put`, `remember`, the decorator, ...) works tagged. It wraps whatever store the repository has, so `Cache.memo().tags(...)` and `Cache.store("file").tags(...)` work too. An empty tag list raises `ValueError`. A tagged repository has no `lock`. + - `TaggedStore` implements the store contract over the wrapped store. `get`, `put` and `forget` use the key `::` (`:` without a prefix). The prefix comes first, as in Laravel, so a store that flushes by prefix, such as the Redis store's `SCAN :*`, still reaches tagged entries. `flush` resets the tags instead of flushing the store. + - Tag ids share the user key space: `tag::key` is an ordinary cache key, so `Cache.put("tag:people:key", ...)` or `Cache.forget(...)` on it changes or resets that tag. Avoid user keys of that form. +- `Repository.tags(names)` accepts a string or a sequence of strings and returns an unprefixed `Repository` over `TaggedStore`, which applies the prefix itself, so every repository method (`get`, `put`, `remember`, the decorator, ...) works tagged. It wraps whatever store the repository has, so `Cache.memo().tags(...)` and `Cache.store("file").tags(...)` work too. An empty tag list raises `ValueError`. A tagged repository has no `lock` and raises `TypeError`. - Flushed entries are not deleted; they stay in the store until they expire or the store is flushed. Use a TTL on tagged entries in long-lived stores. If two workers create a tag's first id at the same time, the last write wins, and entries the other worker wrote under its id become misses. That only costs a recompute. - Reading a tagged entry costs one extra store read per tag. - The facade reaches `tags` through the manager; the stub adds `tags`. @@ -38,4 +39,4 @@ await Cache.tags("people").flush() ## Validation -`tests/cache/test_tagged_store.py` (unittest, array and file stores): tagged put/get round-trips and untagged or differently tagged reads miss; tag order and duplicates do not matter; flushing one tag invalidates every entry carrying it and keeps entries without it, and keeps untagged entries; tag ids persist in the store and are prefixed; forget removes only the tagged key; `remember` and the decorator work tagged; a string tag is accepted; an empty list raises; tags work on a memo repository. +`tests/cache/test_tagged_store.py` (unittest, array, file and in-memory SQLite database stores): the tagged key is `::`, tagged put/get round-trips and untagged or differently tagged reads miss; tag order and duplicates do not matter; flushing one tag invalidates every entry carrying it and keeps entries without it, and keeps untagged entries; tag ids persist in the store and are prefixed; forget removes only the tagged key; `remember` and the decorator work tagged; a string tag is accepted; an empty list raises; `tags().lock()` raises `TypeError`; tags work on a memo repository. diff --git a/docs/adr/AGENTS.md b/docs/adr/AGENTS.md index f69299bc..7fe0e921 100644 --- a/docs/adr/AGENTS.md +++ b/docs/adr/AGENTS.md @@ -34,4 +34,4 @@ Read this index first, then open the records relevant to the change. Before impl | [029](029-cache-module.md) | 2026-10-21 | Cache module with callable remember and pluggable stores | New `fastapi_startkit.cache`: async `Cache.get/put/has/forget/flush`, `get` with a callable default, and `remember(key, ttl, factory)` that runs a sync or async factory only on a miss; `remember(key, ttl)` without a factory is a decorator keyed by `key` plus a hash of the call arguments. Stores implement an async contract with a `MISSING` sentinel (cached `None` is a hit): array, file (pickled per-key files) and database (masoniteorm table). TTL in seconds or `timedelta`, `None` forever, <=0 forgets. `CacheConfig` (`CACHE_CONNECTION`, `CACHE_PREFIX`, `stores`) via `resolve_config`; `CacheProvider` binds `cache`; facade `remember` resolves lazily. A `with` form is impossible in Python. Shipped in slices into `feature/caches`. | | [030](030-cache-memo.md) | 2026-10-21 | Request-scoped cache memoization with Cache.memo() | `Cache.memo(name=None)` returns a `Repository` over `MemoStore`, which wraps the named store and memoizes reads (misses too) for the current asyncio task; `put`/`forget` write through and drop the key, `flush` clears memo and store. State is a `ContextVar` holding `(weakref to owning task, memo)`, and a task that finds another task's memo starts a fresh one, so requests, jobs and child tasks never share memo state. Memoized values ignore TTL for the rest of the task. | | [031](031-cache-locks.md) | 2026-10-21 | Atomic cache locks with Cache.lock() | `Cache.lock(name, seconds=0, owner=None)` returns a store-specific `Lock` (abstract base in `cache/lock.py`): `get(callback=None)` tries once, `block(seconds, callback=None)` polls every 250 ms then raises `LockTimeoutError`, `async with` acquires once or raises, `release()` is owner-checked, `force_release()` is not; callbacks release on error; passing `owner` restores a lock elsewhere; not re-entrant; `seconds<=0` never expires. Array: dict on the store, atomic in the event loop. File: `/locks/` JSON `[owner, expires_at]` under `fcntl.flock`, truncated never unlinked, kept by `flush`, POSIX-only. Database: row `lock:` in the cache table, insert then update-if-expired on conflict, release deletes by owner; no new migration, `flush` clears it. Redis lock follows on the Redis store. | -| [034](034-cache-tags.md) | 2026-10-21 | Tagged cache entries with Cache.tags() | `Cache.tags(name or names)` returns a `Repository` over `TaggedStore`, using Laravel's namespaced-key design. Each tag has a random id stored forever at `:tag::key`. A tagged entry's key is `sha1(ids joined by "|"):`, and `flush()` writes new ids, so every entry carrying any flushed tag becomes unreachable on every store with plain get/put. Tags are deduplicated and sorted, so order does not matter. Flushed entries stay until they expire or the store is flushed. An empty list raises `ValueError`. It composes with `memo()` and named stores. | +| [034](034-cache-tags.md) | 2026-10-21 | Tagged cache entries with Cache.tags() | `Cache.tags(name or names)` returns a `Repository` over `TaggedStore`, using Laravel's namespaced-key design. Each tag has a random id stored forever at `:tag::key`. A tagged entry's key is `:sha1(ids joined by "|"):` (prefix first, so prefix-scoped flushes such as Redis `SCAN` still reach it), tag ids share the user key space, and `flush()` writes new ids, so every entry carrying any flushed tag becomes unreachable on every store with plain get/put. Tags are deduplicated and sorted, so order does not matter. Flushed entries stay until they expire or the store is flushed. An empty list raises `ValueError`. It composes with `memo()` and named stores. | diff --git a/fastapi_startkit/src/fastapi_startkit/cache/repository.py b/fastapi_startkit/src/fastapi_startkit/cache/repository.py index 5cf9b623..67d1e3b1 100644 --- a/fastapi_startkit/src/fastapi_startkit/cache/repository.py +++ b/fastapi_startkit/src/fastapi_startkit/cache/repository.py @@ -121,4 +121,4 @@ def tags(self, names: str | Sequence[str]) -> "Repository": tags = TagSet(self.store, [names] if isinstance(names, str) else names, self.key) if not tags.names: raise ValueError("Cache.tags requires at least one tag.") - return Repository(TaggedStore(self.store, tags), self.prefix) + return Repository(TaggedStore(self.store, tags)) diff --git a/fastapi_startkit/src/fastapi_startkit/cache/tagged_store.py b/fastapi_startkit/src/fastapi_startkit/cache/tagged_store.py index d1d5fa61..fa289e56 100644 --- a/fastapi_startkit/src/fastapi_startkit/cache/tagged_store.py +++ b/fastapi_startkit/src/fastapi_startkit/cache/tagged_store.py @@ -41,7 +41,7 @@ def __init__(self, store: Store, tags: TagSet): self.tags = tags async def tagged(self, key: str) -> str: - return f"{await self.tags.namespace()}:{key}" + return self.tags.key(f"{await self.tags.namespace()}:{key}") async def get(self, key: str) -> Any: return await self.store.get(await self.tagged(key)) diff --git a/fastapi_startkit/tests/cache/test_tagged_store.py b/fastapi_startkit/tests/cache/test_tagged_store.py index 27187148..76a59fbc 100644 --- a/fastapi_startkit/tests/cache/test_tagged_store.py +++ b/fastapi_startkit/tests/cache/test_tagged_store.py @@ -1,15 +1,19 @@ +import hashlib import tempfile -from unittest import IsolatedAsyncioTestCase +from unittest import IsolatedAsyncioTestCase, mock from fastapi_startkit.cache import ArrayStore, CacheManager, FileStore, MemoStore, Repository, TaggedStore +from fastapi_startkit.cache.database_store import DatabaseStore + +from test_database_store import InMemoryDatabase, load_migration class TaggedContract: - def make_store(self): + async def make_store(self): raise NotImplementedError - def setUp(self): - self.store = self.make_store() + async def asyncSetUp(self): + self.store = await self.make_store() self.cache = Repository(self.store, "app") async def test_tagged_put_and_get(self): @@ -94,15 +98,33 @@ async def find(user_id): class TestArrayTags(TaggedContract, IsolatedAsyncioTestCase): - def make_store(self): + async def make_store(self): return ArrayStore() + async def test_tagged_key_starts_with_prefix(self): + await self.cache.tags(["people", "artists"]).put("john", "doe") + artists = await self.cache.get("tag:artists:key") + people = await self.cache.get("tag:people:key") + namespace = hashlib.sha1(f"{artists}|{people}".encode()).hexdigest() + + self.assertIn(f"app:{namespace}:john", self.store.storage) + class TestFileTags(TaggedContract, IsolatedAsyncioTestCase): - def make_store(self): + async def make_store(self): return FileStore(self.enterContext(tempfile.TemporaryDirectory())) +class TestDatabaseTags(TaggedContract, IsolatedAsyncioTestCase): + async def make_store(self): + self.database = InMemoryDatabase() + self.addAsyncCleanup(self.database.engine.dispose) + migration = load_migration() + with mock.patch.object(migration, "config", return_value=None): + await migration.CreateCachesTable("sqlite", self.database.get_schema_builder()).up() + return DatabaseStore(self.database.connection) + + class TestRepositoryTags(IsolatedAsyncioTestCase): async def test_tags_wraps_store_and_keeps_prefix(self): store = ArrayStore() @@ -112,7 +134,19 @@ async def test_tags_wraps_store_and_keeps_prefix(self): self.assertIsInstance(tagged.store, TaggedStore) self.assertIs(tagged.store.store, store) self.assertEqual(tagged.store.tags.names, ["a", "b"]) - self.assertEqual(tagged.prefix, "app") + self.assertEqual(tagged.prefix, "") + + async def test_unprefixed_tagged_key_has_no_leading_separator(self): + store = ArrayStore() + + await Repository(store).tags("people").put("john", "doe") + + self.assertIn("tag:people:key", store.storage) + self.assertEqual(len([key for key in store.storage if key.endswith(":john") and not key.startswith(":")]), 1) + + async def test_tagged_repository_has_no_lock(self): + with self.assertRaises(TypeError): + Repository(ArrayStore(), "app").tags("people").lock("reports") async def test_empty_tags_raise(self): with self.assertRaises(ValueError): From 6b91164c6b1964dc258b5a873b9ff72ca0ee17c1 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Wed, 21 Oct 2026 13:01:13 -0700 Subject: [PATCH 6/6] Run tagged cache contract on Redis and cover store flush of tagged entries --- docs/adr/034-cache-tags.md | 2 +- .../tests/cache/test_tagged_store.py | 21 +++++++++++++++++++ 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/docs/adr/034-cache-tags.md b/docs/adr/034-cache-tags.md index 48f6a309..8ec9ba7d 100644 --- a/docs/adr/034-cache-tags.md +++ b/docs/adr/034-cache-tags.md @@ -39,4 +39,4 @@ await Cache.tags("people").flush() ## Validation -`tests/cache/test_tagged_store.py` (unittest, array, file and in-memory SQLite database stores): the tagged key is `::`, tagged put/get round-trips and untagged or differently tagged reads miss; tag order and duplicates do not matter; flushing one tag invalidates every entry carrying it and keeps entries without it, and keeps untagged entries; tag ids persist in the store and are prefixed; forget removes only the tagged key; `remember` and the decorator work tagged; a string tag is accepted; an empty list raises; `tags().lock()` raises `TypeError`; tags work on a memo repository. +`tests/cache/test_tagged_store.py` (unittest, array, file, in-memory SQLite database and fakeredis Redis stores): the tagged key is `::`, tagged put/get round-trips and untagged or differently tagged reads miss; tag order and duplicates do not matter; flushing one tag invalidates every entry carrying it and keeps entries without it, and keeps untagged entries; tag ids persist in the store and are prefixed; forget removes only the tagged key; `remember` and the decorator work tagged; a string tag is accepted; an empty list raises; `tags().lock()` raises `TypeError`; tags work on a memo repository; a Redis store flush (`SCAN :*`) removes tagged entries and tag ids. diff --git a/fastapi_startkit/tests/cache/test_tagged_store.py b/fastapi_startkit/tests/cache/test_tagged_store.py index 76a59fbc..d2852c3f 100644 --- a/fastapi_startkit/tests/cache/test_tagged_store.py +++ b/fastapi_startkit/tests/cache/test_tagged_store.py @@ -2,8 +2,12 @@ import tempfile from unittest import IsolatedAsyncioTestCase, mock +import fakeredis + from fastapi_startkit.cache import ArrayStore, CacheManager, FileStore, MemoStore, Repository, TaggedStore from fastapi_startkit.cache.database_store import DatabaseStore +from fastapi_startkit.cache.redis_store import RedisStore +from fastapi_startkit.redis import RedisManager from test_database_store import InMemoryDatabase, load_migration @@ -125,6 +129,23 @@ async def make_store(self): return DatabaseStore(self.database.connection) +class TestRedisTags(TaggedContract, IsolatedAsyncioTestCase): + async def make_store(self): + server = fakeredis.FakeServer() + manager = RedisManager( + {"client": "fake", "options": {"prefix": "app_"}, "connections": {"cache": {"database": 1}}} + ).extend("fake", lambda parameters: fakeredis.FakeAsyncRedis(server=server, **parameters)) + return RedisStore(lambda: manager.connection("cache"), "app") + + async def test_store_flush_removes_tagged_entries(self): + await self.cache.tags("people").put("john", "doe") + + await self.cache.flush() + + self.assertEqual([key async for key in self.store.connection().scan_iter(match="app_*")], []) + self.assertIsNone(await self.cache.tags("people").get("john")) + + class TestRepositoryTags(IsolatedAsyncioTestCase): async def test_tags_wraps_store_and_keeps_prefix(self): store = ArrayStore()