diff --git a/docs/adr/029-cache-module.md b/docs/adr/029-cache-module.md new file mode 100644 index 00000000..8434dd37 --- /dev/null +++ b/docs/adr/029-cache-module.md @@ -0,0 +1,162 @@ +--- +title: Cache module with remember, memo, locks and tags +date: 2026-10-21 +--- + +## Problem + +The framework ships a `Cache` facade bound to the `cache` key, but nothing registers that key, so every call fails. Applications need a Laravel-style cache: `get`, `put`, `forget`, `flush`, multiple named stores, and `remember`, which computes a value only on a miss. FastAPI code is async, so the API must be awaitable and must accept both sync and async factories. + +On top of the basic cache, applications need the rest of Laravel's cache features: + +- **Memoization.** A request or job often reads the same key several times (settings, the current tenant). With the file, database or Redis store each read is I/O and unpickling. Laravel's `Cache::memo()` remembers values read from the store for the rest of the request. The framework has no request object to hang that state on, so state must follow the running task without leaking between concurrent requests. +- **Atomic locks.** Only one worker may run some 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. A check-then-write `put` cannot be used, because two workers can both see the key missing. +- **Tags.** Applications cache many related entries (every page of a user's posts, every report for a team) and must invalidate the group without knowing every key. Laravel's `Cache::tags([...])` reads entries only with the same tags, and `Cache::tags([...])->flush()` invalidates every entry that carries any of those tags. + +## Alternatives + +Core API: + +- **`with Cache.remember(key, ttl):` block.** A context manager cannot skip its body or return a value, so it cannot express "compute only on a miss". Rejected. +- **Decorator-only API.** Fits functions, not inline expressions such as `await Cache.remember("users", 60, query)`. Rejected as the only form. +- **Third-party library (aiocache, cashews).** Adds a dependency and its own config model, and does not match the provider/config/facade pattern used by Redis and Storage. Rejected. +- **Sync API with `asyncio.to_thread` wrappers for callers.** Breaks the database store, which is async-only. Rejected. +- **Callable core API plus a decorator built on it, behind a store protocol.** Chosen. + +Memoization: + +- **Process-wide dict.** Shared by every request, so one request sees values another request cached and never sees later writes from other workers. Rejected. +- **Middleware that opens a scope per request.** Needs FastAPI wiring in the cache provider, does nothing for queue jobs or console commands, and a memo used outside the middleware would fall back to something else anyway. Rejected. +- **`ContextVar` holding a dict, created on first use.** A dict created in a parent context (lifespan, a test fixture, a worker loop) is inherited by every child task, so all requests would share it. Rejected on its own. +- **`ContextVar` holding a dict tagged with the task that created it.** A task that finds a dict owned by another task starts a fresh one. Chosen. + +Locks: + +- **Lock built on `get` + `put`.** Not atomic across processes. Rejected. +- **`asyncio.Lock`.** Process-local only. Rejected. +- **One generic lock over 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. + +Tags: + +- **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, and 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 + +### Configuration, stores and repository + +- Package `fastapi_startkit.cache` with `CacheConfig` in `cache/config/__init__.py` (`default`, `prefix`, `stores`) and one config dataclass per driver: `ArrayStoreConfig`, `FileStoreConfig(path)`, `DatabaseStoreConfig(connection, table)` and `RedisStoreConfig(connection)`. Env names: `CACHE_CONNECTION` (default `database`), `CACHE_PREFIX` (default `cache`), `CACHE_FILE_PATH`, `DB_CACHE_CONNECTION` (default: the default database connection) and `DB_CACHE_TABLE` (default `caches`). The provider reads it with `resolve_config`, so a plain dataclass of the same shape works (ADR 027). +- Store contract (async): `get(key)` returns a `MISSING` sentinel on a miss so that a cached `None` is a hit; `put(key, value, seconds | None)`, `forget(key)` and `flush()` return `bool`. +- `ArrayStore`: in-process dict of `(expires_at, value)`, values kept as-is. +- `FileStore`: one pickled `(expires_at, value)` file per key under `//` inside the configured path. Writes are atomic through a temp file and `os.replace` (a failed write removes the temp file and keeps the previous value), and file I/O runs in `asyncio.to_thread`. The provider resolves a relative path against the application base path. Unreadable or corrupt files, including values whose class was renamed or removed, are a miss and deleted. `flush` removes only the two-hex-character shard directories, so a misconfigured path such as `storage` never loses unrelated files. +- `DatabaseStore`: masoniteorm table with `key` string primary key, `value` text holding base64 pickle and nullable big integer `expiration` (epoch seconds, rounded up; a 32-bit column would overflow in 2038). The connection is resolved from the `db` binding on each call, so provider order does not matter. `put` updates first and inserts when no row matched, retrying the update when a concurrent insert wins, because the ORM has no portable upsert. Inside a caller's open PostgreSQL transaction the failed insert aborts that transaction, so the retry cannot recover there and the `IntegrityError` propagates; MySQL and SQLite keep the transaction usable. Corrupt or expired rows, including values whose class was renamed or removed, are a miss and deleted. Queries go through the query builder so the reserved word `key` is quoted per grammar. The module imports SQLAlchemy and is loaded lazily by the provider, so the cache works without the `database` extra. The provider publishes the `create_caches_table` migration to `databases/migrations/`; the migration reads the table name from `cache.stores.database.table` (`DB_CACHE_TABLE`), falling back to `caches`. +- `RedisStore` (driver `redis`, `RedisStoreConfig(connection="cache")`) reuses the existing Redis support instead of creating its own client. It resolves the named connection from the `redis` binding (`RedisManager`, registered by `RedisProvider`) on each call; without `RedisProvider`, resolving the store still succeeds and the first operation raises a `RuntimeError` naming it. It therefore gets the `redis` extra, the `REDIS_*` connection settings, the lazy `redis.asyncio` import and the connection-level `REDIS_PREFIX` for free. `connection` selects any configured Redis connection (`cache` uses `REDIS_CACHE_DB`, `default` the main database). Values are base64 pickle like the database store, so they survive the manager's `decode_responses=True`; corrupt values are a miss and deleted. TTL is `SET ... PX` in milliseconds, rounded up to at least 1 ms, and Redis expires keys itself. `flush` never calls `FLUSHDB`/`FLUSHALL`: it `SCAN`s for `:*` (glob characters escaped), strips the connection prefix that the connection re-applies, and `UNLINK`s in batches of 500. The provider passes the cache `prefix` to the store, and an empty prefix is rejected because flush could not be scoped. +- Pickle allows arbitrary Python values, and unpickling runs code, so the file directory, the cache table and the Redis database are trusted storage: never point them at a location that untrusted parties can write. +- `Repository` wraps one store and adds the key prefix (`:` when a prefix is set) and the public API: `get(key, default=None)` where a callable default is called (and awaited when it returns an awaitable) only on a miss, `put(key, value, ttl=None)`, `has`, `forget`, `flush`, `remember(key, ttl, factory)`, `lock` and `tags`. +- TTL is seconds or a `timedelta`; `None` stores forever; zero or less forgets the key and returns `False`. Expiry is an absolute epoch timestamp checked lazily on read, and expired entries are deleted when read. +- `remember(key, ttl)` without a factory returns a decorator. The decorated function becomes async; the cache key is `key` alone for a call without arguments, otherwise `key:` plus a SHA-256 prefix of the pickled `(args, sorted kwargs)` (falling back to `repr` for unpicklable arguments). Arguments that cannot be pickled and whose `repr` is not stable (for example one that includes a memory address that changes per call) produce a new key every call and always miss. `set` and `frozenset` arguments pickle in iteration order, which can differ for equal sets built in a different order or in another process (string hashing is randomized per process), so pass a sorted tuple instead. A `None` key uses `module.qualname` of the function; the factory form requires a key and raises `ValueError` without one. The decorator and the factory form share entries, so `put("users", ...)` is seen by a decorated function keyed `users` that takes no arguments. Instance methods include `self` in the hashed arguments, so each instance caches separately. +- `CacheManager` resolves a `Repository` per store name, caches it, delegates unknown attributes to the default store, and supports `extend(driver, factory)` for custom stores. `CacheProvider` merges config under `cache`, binds the manager as `cache` and publishes `config/cache.py`. The `Cache` facade defines `remember` on the facade class so decorating at import time does not resolve the container; resolution happens per call. A `.pyi` stub types the facade, including `memo`, `lock` and `tags`. + +### Memoization: `Cache.memo()` + +- `cache/memo_store.py` adds `MemoStore(store)`, a store that wraps another store. `get` returns the memoized value for the key when present, otherwise reads the inner store and memoizes the result, including `MISSING`, so repeated misses are free too. `put` and `forget` write through and drop the memoized key, so the next `get` re-reads the store. `flush` flushes the inner store and clears this store's memo. +- Memo state lives in one module `ContextVar` holding `(weakref to owning task, {MemoStore: {key: value}})`. On access, when there is no state or the owner is not `asyncio.current_task()`, a new dict is created and set for the current context. Each request, job or `asyncio` task therefore starts empty, and a dict is never shared with a sibling task. Child tasks (`asyncio.gather`, `create_task`) start their own memo and still read through to the same store. +- Memoized values ignore TTL for the rest of the task: a value that expires in the store mid-request stays visible to that request, matching Laravel. Values are kept as the same object, so mutating a memoized value mutates what later reads in the task see. +- `CacheManager.memo(name=None)` returns a `Repository` over `MemoStore(store(name).store)` with the same prefix, cached per store name, so `Cache.memo()` uses the default store and `Cache.memo("file")` a named one. Every repository method works through it. + +### Atomic locks: `Cache.lock()` + +- `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 it has not expired, and returns `bool`; releasing 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 shard directories, so `flush` keeps them. `fcntl` is imported on use, so the file cache still imports on Windows; file store locks 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 `put` applies to the insert conflict. + - `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 an own expired lock returns `False`. The key is outside the store's `:` namespace, so `flush` keeps Redis locks, and it 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 (memo and tagged repositories) raises `TypeError`. + +### Tags: `Cache.tags()` + +- `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 `::` (`:` 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 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`. +- Flushed entries are not deleted; they stay in the store until they expire or the store is flushed, so 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, which only costs a recompute. Reading a tagged entry costs one extra store read per tag. + +## Usage + +```python +from fastapi_startkit.facades import Cache + +value = await Cache.get("key") +value = await Cache.get("key", "default") +value = await Cache.get("key", lambda: compute_default()) + +await Cache.put("key", "value", 600) +await Cache.put("key", "value", timedelta(minutes=10)) +await Cache.forever("key", "value") +await Cache.forget("key") +await Cache.flush() + +value = await Cache.store("redis").get("key") + +users = await Cache.remember("users", 600, lambda: DB.table("users").get()) + + +@Cache.remember("users", 600) +async def users_by_role(role: str): + return await DB.table("users").where("role", role).get() + + +settings = await Cache.memo().get("settings") +tenant = await Cache.memo("redis").remember("tenant", 600, load_tenant) + +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() + +await Cache.tags(["people", "artists"]).put("john", john, 600) +john = await Cache.tags(["people", "artists"]).get("john") +await Cache.tags("people").flush() +``` + +The default `database` store needs the `caches` table: run `artisan provider:publish` to copy the migration into `databases/migrations/`, then `artisan db:migrate`. Set `CACHE_CONNECTION=file` to skip the table, or `CACHE_CONNECTION=redis` (with `RedisProvider` registered) to use Redis. + +A plain dataclass with the same shape configures it: + +```python +@dataclass +class CacheConfig: + default: str = env("CACHE_CONNECTION", "database") + prefix: str = env("CACHE_PREFIX", "cache") + stores: dict = field( + default_factory=lambda: { + "array": ArrayStoreConfig(), + "file": FileStoreConfig(path="storage/framework/cache/data"), + "database": DatabaseStoreConfig(table="caches"), + "redis": RedisStoreConfig(connection="cache"), + } + ) +``` + +## Delivery + +Shipped as small PRs into the `feature/caches` integration branch: config; store contract, `ArrayStore`, `Repository` and `CacheManager`; the `remember` decorator; provider, facade and stub; `FileStore`; `DatabaseStore` and migration; `RedisStore`; `Cache.memo()`; `Cache.lock()` for array, file and database; the Redis lock; `Cache.tags()`. The memo, lock and tags work, first recorded as ADRs 030, 031 and 034, is folded into this record. + +## Validation + +Unittest suites under `fastapi_startkit/tests/cache/`: + +- Config defaults and env overrides. Per store: miss/hit, cached `None`, expiry, forever, non-positive TTL, forget, flush. `get` with sync and async callable defaults; `remember` runs the factory once and awaits async factories; decorator keys differ per argument set and the function runs once per key. Provider binding, facade access after boot and the database store as default. +- Database store: upsert under a concurrent insert, corrupt rows, values of removed classes, the big integer expiration, and the migration up/down and configured table name on SQLite. Redis store: round trip, millisecond TTL, connection prefix and selectable connection, flush deleting only prefixed keys in batches with glob characters escaped and without `FLUSHDB`, using `fakeredis` through `RedisManager.extend`. +- `test_memo_store.py`: a repeated `get` reads the inner store once; a miss is memoized; `put`/`forget` invalidate and write through; `flush` clears memo and store; separate tasks do not share memo state, including when state exists in the parent context; two stores keep separate memos; the manager returns one memo repository per store with the configured prefix. +- `test_lock.py`: 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, 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, `async with` acquires, releases when the body raises, and raises when held. The file store checks that `flush` keeps locks and a corrupt lock file counts as free; Redis checks the prefixed key and TTL, no collision with a cache key, and that `flush` keeps locks. The repository prefixes lock names, and a store without locks raises `TypeError`. +- `test_tagged_store.py` (array, file, in-memory SQLite and fakeredis): the tagged key is `::`; tagged 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 other and untagged entries; tag ids persist 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 removes tagged entries and tag ids. diff --git a/docs/adr/AGENTS.md b/docs/adr/AGENTS.md index b65dfe83..fcedd5b6 100644 --- a/docs/adr/AGENTS.md +++ b/docs/adr/AGENTS.md @@ -32,4 +32,5 @@ Read this index first, then open the records relevant to the change. Before impl | [024](024-installer-controller-module.md) | 2026-10-19 | Installer controllers in their own module | `fastapi-startkit init` generates `HomeController` in `app/http/controllers/home_controller.py` (JSON or Inertia variant) instead of inside `routes/api.py`; one shared routes stub imports and registers it, matching the PingCRM layout. | | [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. | | [026](026-schema-column-change.md) | 2026-10-20 | Altering existing columns with `.change()` | `.change()` fully redefines the column, the way Laravel does. MySQL (`MODIFY`) and SQLite (table rebuild) were already correct. Postgres now emits `TYPE ... USING` casts, correct lengths, quoted defaults or `DROP DEFAULT`, `SET`/`DROP NOT NULL`, and replaces the enum check constraint. MSSQL drops the current default constraint (looked up in `sys.default_constraints`), runs one `ALTER COLUMN` per column, then adds `[DF__]`. Identifiers are bracket- or double-quote-escaped and the MSSQL drop target is built server-side with `QUOTENAME`; changing a Postgres serial column drops its `nextval` default. | +| [029](029-cache-module.md) | 2026-10-21 | Cache module with remember, memo, locks and tags | 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` via `resolve_config`; `CacheProvider` binds `cache`; facade `remember` resolves lazily. `Cache.memo()` memoizes reads per asyncio task through a `ContextVar` owned by the task, so requests never share memo state. `Cache.lock(name, seconds, owner)` gives atomic, owner-checked, non-re-entrant locks with `get`, `block` (250 ms polling, `LockTimeoutError`), `release`, `force_release` and `async with`, per store: event-loop dict, `fcntl.flock` file, `lock:` row with update-if-expired, Redis `SET NX PX` plus a Lua owner-checked delete. `Cache.tags([...])` uses Laravel's namespaced keys: random tag ids at `:tag::key`, entries at `:sha1(ids):`, flush rewrites the ids; tags are sorted and deduplicated. Shipped in slices into `feature/caches`; the former memo, lock and tags ADRs (030, 031, 034) are folded in. | | [032](032-migrate-seed-option.md) | 2026-10-21 | db:migrate:fresh and db:migrate:refresh --seed run seeders in the same event loop | `--seed` called the unregistered `seed:run` through Cleo `call` (which also drops the first argument) inside a running event loop, where `db:seed` would call `asyncio.run` again. The seeder selection moves into async `DBSeedCommand.run_seeders`, which `db:seed`, `db:migrate:fresh` and `db:migrate:refresh` all await. `db:migrate:refresh` is now registered. | diff --git a/fastapi_startkit/src/fastapi_startkit/cache/__init__.py b/fastapi_startkit/src/fastapi_startkit/cache/__init__.py new file mode 100644 index 00000000..ad1bbb29 --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/cache/__init__.py @@ -0,0 +1,30 @@ +from .array_store import ArrayStore +from .config import ArrayStoreConfig, CacheConfig, DatabaseStoreConfig, FileStoreConfig, RedisStoreConfig +from .file_store import FileStore +from .lock import Lock, LockTimeoutError +from .manager import CacheManager +from .memo_store import MemoStore +from .provider import CacheProvider +from .repository import Repository +from .store import MISSING, Store +from .tagged_store import TaggedStore, TagSet + +__all__ = [ + "MISSING", + "ArrayStore", + "ArrayStoreConfig", + "CacheConfig", + "CacheManager", + "CacheProvider", + "DatabaseStoreConfig", + "FileStore", + "FileStoreConfig", + "Lock", + "LockTimeoutError", + "MemoStore", + "RedisStoreConfig", + "Repository", + "Store", + "TagSet", + "TaggedStore", +] diff --git a/fastapi_startkit/src/fastapi_startkit/cache/array_store.py b/fastapi_startkit/src/fastapi_startkit/cache/array_store.py new file mode 100644 index 00000000..05a85338 --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/cache/array_store.py @@ -0,0 +1,62 @@ +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) + if entry is None: + return MISSING + + expiration, value = entry + if expired(expiration): + del self.storage[key] + return MISSING + return value + + async def put(self, key: str, value: Any, seconds: float | None) -> bool: + self.storage[key] = (expires_at(seconds), value) + return True + + async def forget(self, key: str) -> bool: + return self.storage.pop(key, MISSING) is not MISSING + + 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/config/__init__.py b/fastapi_startkit/src/fastapi_startkit/cache/config/__init__.py new file mode 100644 index 00000000..5bd18cee --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/cache/config/__init__.py @@ -0,0 +1,45 @@ +from dataclasses import dataclass, field + +from fastapi_startkit.configuration import Config +from fastapi_startkit.environment import env + + +@dataclass +class ArrayStoreConfig(Config): + driver: str = "array" + + +@dataclass +class FileStoreConfig(Config): + driver: str = "file" + path: str = env("CACHE_FILE_PATH", "storage/framework/cache/data") + + +@dataclass +class DatabaseStoreConfig(Config): + driver: str = "database" + connection: str | None = env("DB_CACHE_CONNECTION", None, cast=False) + table: str = env("DB_CACHE_TABLE", "caches") + + +@dataclass +class RedisStoreConfig(Config): + driver: str = "redis" + connection: str = "cache" + + +@dataclass +class CacheConfig(Config): + default: str = env("CACHE_CONNECTION", "database") + prefix: str = env("CACHE_PREFIX", "cache") + stores: dict[str, Config] = field( + default_factory=lambda: { + "array": ArrayStoreConfig(), + "file": FileStoreConfig(), + "database": DatabaseStoreConfig(), + "redis": RedisStoreConfig(), + } + ) + + +__all__ = ["ArrayStoreConfig", "CacheConfig", "DatabaseStoreConfig", "FileStoreConfig", "RedisStoreConfig"] diff --git a/fastapi_startkit/src/fastapi_startkit/cache/database_store.py b/fastapi_startkit/src/fastapi_startkit/cache/database_store.py new file mode 100644 index 00000000..d182c5f2 --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/cache/database_store.py @@ -0,0 +1,94 @@ +import base64 +import math +import pickle +import time +from collections.abc import Callable +from typing import Any + +from sqlalchemy.exc import IntegrityError + +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 + self.table = table + + def query(self) -> QueryBuilder[Any]: + return self.connection().query().table(self.table) + + async def get(self, key: str) -> Any: + builder = self.query().where("key", key).limit(1) + row = await builder.connection.select_one(builder.to_qmark(), list(builder.get_bindings())) + if row is None: + return MISSING + + try: + value = pickle.loads(base64.b64decode(row["value"])) + except Exception: + await self.forget(key) + return MISSING + + if expired(row["expiration"]): + await self.forget(key) + return MISSING + return value + + async def put(self, key: str, value: Any, seconds: float | None) -> bool: + expiration = expires_at(seconds) + 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: + await self.query().insert({"key": key, **record}) + except IntegrityError: + await self.query().where("key", key).update(record) + return True + + async def forget(self, key: str) -> bool: + return bool(await self.query().where("key", key).delete()) + + 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: + 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/src/fastapi_startkit/cache/file_store.py b/fastapi_startkit/src/fastapi_startkit/cache/file_store.py new file mode 100644 index 00000000..fda1e33e --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/cache/file_store.py @@ -0,0 +1,135 @@ +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 .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: + def __init__(self, directory: str | Path): + self.directory = Path(directory) + + def path(self, key: str) -> Path: + digest = hashlib.sha1(key.encode()).hexdigest() + return self.directory / digest[:2] / digest[2:4] / digest + + async def get(self, key: str) -> Any: + return await asyncio.to_thread(self.read, self.path(key)) + + async def put(self, key: str, value: Any, seconds: float | None) -> bool: + payload = pickle.dumps((expires_at(seconds), value)) + await asyncio.to_thread(self.write, self.path(key), payload) + return True + + async def forget(self, key: str) -> bool: + return await asyncio.to_thread(self.delete, self.path(key)) + + async def flush(self) -> bool: + await asyncio.to_thread(self.clear) + return True + + def read(self, path: Path) -> Any: + try: + payload = path.read_bytes() + except FileNotFoundError: + return MISSING + try: + expiration, value = pickle.loads(payload) + except Exception: + self.delete(path) + return MISSING + + if expired(expiration): + self.delete(path) + return MISSING + return value + + def write(self, path: Path, payload: bytes) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary = tempfile.mkstemp(dir=path.parent, prefix=".tmp-") + try: + with os.fdopen(descriptor, "wb") as file: + file.write(payload) + os.replace(temporary, path) + except BaseException: + Path(temporary).unlink(missing_ok=True) + raise + + def delete(self, path: Path) -> bool: + try: + path.unlink() + except FileNotFoundError: + return False + return True + + def clear(self) -> None: + if not self.directory.is_dir(): + return + 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/manager.py b/fastapi_startkit/src/fastapi_startkit/cache/manager.py new file mode 100644 index 00000000..0c1642b6 --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/cache/manager.py @@ -0,0 +1,49 @@ +from collections.abc import Callable +from typing import Any, Self + +from .array_store import ArrayStore +from .memo_store import MemoStore +from .repository import Repository +from .store import Store + +StoreFactory = Callable[[dict[str, Any]], Store] + + +class CacheManager: + def __init__(self, config: dict[str, Any]): + self.config = config + self._stores: dict[str, Repository] = {} + self._memos: dict[str, Repository] = {} + self._creators: dict[str, StoreFactory] = {"array": lambda config: ArrayStore()} + + def __getattr__(self, method: str) -> Any: + if method.startswith("_"): + raise AttributeError(method) + return getattr(self.store(), method) + + def store(self, name: str | None = None) -> Repository: + resolved: str = name or self.config.get("default") or "array" + if resolved not in self._stores: + self._stores[resolved] = self.resolve(resolved) + return self._stores[resolved] + + def memo(self, name: str | None = None) -> Repository: + resolved: str = name or self.config.get("default") or "array" + if resolved not in self._memos: + repository = self.store(resolved) + self._memos[resolved] = Repository(MemoStore(repository.store), repository.prefix) + return self._memos[resolved] + + def resolve(self, name: str) -> Repository: + config = self.config.get("stores", {}).get(name) + if config is None: + raise ValueError(f"Cache store [{name}] is not defined.") + + driver = config.get("driver") + if driver not in self._creators: + raise ValueError(f"Cache driver [{driver}] is not supported.") + return Repository(self._creators[driver](config), self.config.get("prefix") or "") + + def extend(self, driver: str, factory: StoreFactory) -> Self: + self._creators[driver] = factory + return self diff --git a/fastapi_startkit/src/fastapi_startkit/cache/memo_store.py b/fastapi_startkit/src/fastapi_startkit/cache/memo_store.py new file mode 100644 index 00000000..045fc494 --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/cache/memo_store.py @@ -0,0 +1,48 @@ +import asyncio +import weakref +from contextvars import ContextVar +from typing import Any + +from .store import Store + +Memo = dict["MemoStore", dict[str, Any]] + +_scope: ContextVar[tuple[weakref.ref[asyncio.Task[Any]], Memo] | None] = ContextVar("cache_memo", default=None) + + +def memo() -> Memo: + task = asyncio.current_task() + state = _scope.get() + if state is not None and task is not None and state[0]() is task: + return state[1] + + fresh: Memo = {} + if task is not None: + _scope.set((weakref.ref(task), fresh)) + return fresh + + +class MemoStore: + def __init__(self, store: Store): + self.store = store + + def values(self) -> dict[str, Any]: + return memo().setdefault(self, {}) + + async def get(self, key: str) -> Any: + values = self.values() + if key not in values: + values[key] = await self.store.get(key) + return values[key] + + async def put(self, key: str, value: Any, seconds: float | None) -> bool: + self.values().pop(key, None) + return await self.store.put(key, value, seconds) + + async def forget(self, key: str) -> bool: + self.values().pop(key, None) + return await self.store.forget(key) + + async def flush(self) -> bool: + self.values().clear() + return await self.store.flush() diff --git a/fastapi_startkit/src/fastapi_startkit/cache/migrations/2026_10_21_000000_create_caches_table.py b/fastapi_startkit/src/fastapi_startkit/cache/migrations/2026_10_21_000000_create_caches_table.py new file mode 100644 index 00000000..c7d6167f --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/cache/migrations/2026_10_21_000000_create_caches_table.py @@ -0,0 +1,17 @@ +from fastapi_startkit.configuration import config +from fastapi_startkit.masoniteorm.migrations import Migration + + +def table_name() -> str: + return config("cache.stores.database.table") or "caches" + + +class CreateCachesTable(Migration): + async def up(self): + async with await self.schema.create(table_name()) as table: + table.string("key").primary() + table.text("value") + table.big_integer("expiration").nullable() + + async def down(self): + await self.schema.drop(table_name()) diff --git a/fastapi_startkit/src/fastapi_startkit/cache/provider.py b/fastapi_startkit/src/fastapi_startkit/cache/provider.py new file mode 100644 index 00000000..7cd98245 --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/cache/provider.py @@ -0,0 +1,48 @@ +from pathlib import Path +from typing import Any + +from fastapi_startkit.support import Provider + +from .config import CacheConfig +from .file_store import FileStore +from .manager import CacheManager +from .store import Store + +MIGRATION = "2026_10_21_000000_create_caches_table.py" + + +class CacheProvider(Provider): + provider_key = "cache" + + def register(self) -> None: + config = self.resolve_config(CacheConfig) + self.merge_config_from(config, self.provider_key) + manager = CacheManager(config) + manager.extend("file", lambda store: FileStore(Path(self.app.base_path) / store["path"])) + manager.extend("database", self.database_store) + manager.extend("redis", lambda store: self.redis_store(store, config.get("prefix") or "")) + self.app.bind("cache", manager) + + def database_store(self, store: dict[str, Any]) -> Store: + from .database_store import DatabaseStore + + return DatabaseStore(lambda: self.app.make("db").connection(store.get("connection")), store["table"]) + + def redis_store(self, store: dict[str, Any], prefix: str) -> Store: + from .redis_store import RedisStore + + return RedisStore(lambda: self.redis().connection(store.get("connection")), prefix) + + def redis(self) -> Any: + if not self.app.has("redis"): + raise RuntimeError("The redis cache store requires RedisProvider to be registered.") + return self.app.make("redis") + + def boot(self) -> None: + directory = Path(__file__).resolve().parent + self.publishes( + { + directory / "config" / "__init__.py": "config/cache.py", + directory / "migrations" / MIGRATION: f"databases/migrations/{MIGRATION}", + } + ) diff --git a/fastapi_startkit/src/fastapi_startkit/cache/redis_store.py b/fastapi_startkit/src/fastapi_startkit/cache/redis_store.py new file mode 100644 index 00000000..f923248f --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/cache/redis_store.py @@ -0,0 +1,85 @@ +import base64 +import math +import pickle +import re +from collections.abc import Callable +from typing import Any + +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: + def __init__(self, connection: Callable[[], Connection], prefix: str, batch: int = 500): + if not prefix: + raise ValueError("Redis cache store requires a non-empty prefix.") + self.connection = connection + self.prefix = prefix + self.batch = batch + + async def get(self, key: str) -> Any: + payload = await self.connection().get(key) + if payload is None: + return MISSING + try: + return pickle.loads(base64.b64decode(payload)) + except Exception: + await self.forget(key) + return MISSING + + async def put(self, key: str, value: Any, seconds: float | None) -> bool: + payload = base64.b64encode(pickle.dumps(value)).decode() + milliseconds = None if seconds is None else max(1, math.ceil(seconds * 1000)) + return bool(await self.connection().set(key, payload, px=milliseconds)) + + async def forget(self, key: str) -> bool: + return bool(await self.connection().delete(key)) + + async def flush(self) -> bool: + connection = self.connection() + namespace = f"{connection.prefix}{self.prefix}:" + pattern = GLOB.sub(r"\\\1", namespace) + "*" + keys: list[Any] = [] + async for key in connection.scan_iter(match=pattern, count=self.batch): + keys.append(strip(key, connection.prefix)) + if len(keys) >= self.batch: + await connection.unlink(*keys) + keys.clear() + if keys: + 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 + return key[len(marker) :] diff --git a/fastapi_startkit/src/fastapi_startkit/cache/repository.py b/fastapi_startkit/src/fastapi_startkit/cache/repository.py new file mode 100644 index 00000000..67d1e3b1 --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/cache/repository.py @@ -0,0 +1,124 @@ +import functools +import hashlib +import inspect +import pickle +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 + +P = ParamSpec("P") +Ttl = float | timedelta | None + + +class Decorator(Protocol): + def __call__(self, function: Callable[P, Any], /) -> Callable[P, Awaitable[Any]]: ... + + +def call_key(base: str, args: tuple[Any, ...], kwargs: dict[str, Any]) -> str: + if not args and not kwargs: + return base + + arguments = (args, sorted(kwargs.items())) + try: + payload = pickle.dumps(arguments) + except (pickle.PicklingError, TypeError, AttributeError): + payload = repr(arguments).encode() + return f"{base}:{hashlib.sha256(payload).hexdigest()[:16]}" + + +async def evaluate(value: Any) -> Any: + if callable(value): + value = value() + if inspect.isawaitable(value): + value = await value + return value + + +def seconds(ttl: Ttl) -> float | None: + return ttl.total_seconds() if isinstance(ttl, timedelta) else ttl + + +class Repository: + def __init__(self, store: Store, prefix: str = ""): + self.store = store + self.prefix = prefix + + def key(self, key: str) -> str: + return f"{self.prefix}:{key}" if self.prefix else key + + async def get(self, key: str, default: Any = None) -> Any: + value = await self.store.get(self.key(key)) + return await evaluate(default) if value is MISSING else value + + async def has(self, key: str) -> bool: + return await self.store.get(self.key(key)) is not MISSING + + async def put(self, key: str, value: Any, ttl: Ttl = None) -> bool: + duration = seconds(ttl) + if duration is not None and duration <= 0: + await self.forget(key) + return False + return await self.store.put(self.key(key), value, duration) + + async def forever(self, key: str, value: Any) -> bool: + return await self.put(key, value) + + @overload + 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 remember( + self, key: str | None, ttl: Ttl, factory: Callable[[], Any] | None = None + ) -> Awaitable[Any] | Decorator: + if factory is None: + return self.decorator(key, ttl) + if key is None: + raise ValueError("Cache.remember requires a key when a factory is given.") + return self.fetch(key, ttl, factory) + + def decorator(self, key: str | None, ttl: Ttl) -> Decorator: + def decorate(function: Callable[P, Any]) -> Callable[P, Awaitable[Any]]: + base = key or f"{function.__module__}.{function.__qualname__}" + + @functools.wraps(function) + async def cached(*args: P.args, **kwargs: P.kwargs) -> Any: + return await self.fetch(call_key(base, args, kwargs), ttl, lambda: function(*args, **kwargs)) + + return cached + + return decorate + + async def fetch(self, key: str, ttl: Ttl, factory: Callable[[], Any]) -> Any: + value = await self.store.get(self.key(key)) + if value is not MISSING: + return value + + value = await evaluate(factory) + await self.put(key, value, ttl) + return value + + async def forget(self, key: str) -> bool: + return await self.store.forget(self.key(key)) + + 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) + + 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)) diff --git a/fastapi_startkit/src/fastapi_startkit/cache/store.py b/fastapi_startkit/src/fastapi_startkit/cache/store.py new file mode 100644 index 00000000..171bc0f3 --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/cache/store.py @@ -0,0 +1,29 @@ +import time +from typing import Any, Final, Protocol, final + + +@final +class Missing: + def __repr__(self) -> str: + return "MISSING" + + +MISSING: Final = Missing() + + +class Store(Protocol): + async def get(self, key: str) -> Any: ... + + async def put(self, key: str, value: Any, seconds: float | None) -> bool: ... + + async def forget(self, key: str) -> bool: ... + + async def flush(self) -> bool: ... + + +def expires_at(seconds: float | None) -> float | None: + return None if seconds is None else time.time() + seconds + + +def expired(expiration: float | None) -> bool: + return expiration is not None and expiration <= time.time() 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..fa289e56 --- /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 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)) + + 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.py b/fastapi_startkit/src/fastapi_startkit/facades/Cache.py index d662862e..6d39bee9 100644 --- a/fastapi_startkit/src/fastapi_startkit/facades/Cache.py +++ b/fastapi_startkit/src/fastapi_startkit/facades/Cache.py @@ -1,5 +1,29 @@ +import functools +from collections.abc import Callable +from typing import Any + from .Facade import Facade +def manager() -> Any: + from ..application import app + + return app().make("cache") + + class Cache(metaclass=Facade): key = "cache" + + @staticmethod + def remember(key: str | None, ttl: Any, factory: Callable[[], Any] | None = None) -> Any: + if factory is not None: + return manager().remember(key, ttl, factory) + + def decorate(function: Callable[..., Any]) -> Callable[..., Any]: + @functools.wraps(function) + async def cached(*args: Any, **kwargs: Any) -> Any: + return await manager().remember(key, ttl)(function)(*args, **kwargs) + + return cached + + return decorate diff --git a/fastapi_startkit/src/fastapi_startkit/facades/Cache.pyi b/fastapi_startkit/src/fastapi_startkit/facades/Cache.pyi new file mode 100644 index 00000000..aad7cc9e --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/facades/Cache.pyi @@ -0,0 +1,25 @@ +from collections.abc import Awaitable, Callable, Sequence +from typing import Any, overload + +from fastapi_startkit.cache import CacheManager, Lock, Repository +from fastapi_startkit.cache.manager import StoreFactory +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: ... + async def put(self, key: str, value: Any, ttl: Ttl = None) -> bool: ... + async def forever(self, key: str, value: Any) -> bool: ... + @overload + 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: ... + +Cache: CacheFacade diff --git a/fastapi_startkit/tests/cache/test_array_store.py b/fastapi_startkit/tests/cache/test_array_store.py new file mode 100644 index 00000000..60b5948a --- /dev/null +++ b/fastapi_startkit/tests/cache/test_array_store.py @@ -0,0 +1,54 @@ +from unittest import IsolatedAsyncioTestCase, mock + +from fastapi_startkit.cache import MISSING, ArrayStore + + +class TestArrayStore(IsolatedAsyncioTestCase): + def setUp(self): + self.time = self.enterContext(mock.patch("time.time", return_value=1_000_000.0)) + self.store = ArrayStore() + + async def test_missing_key_returns_sentinel(self): + self.assertIs(await self.store.get("name"), MISSING) + + async def test_put_and_get_keep_the_same_object(self): + value = {"roles": ["admin"]} + + self.assertIs(await self.store.put("user", value, 60), True) + self.assertIs(await self.store.get("user"), value) + + async def test_cached_none_is_a_hit(self): + await self.store.put("nothing", None, None) + + self.assertIsNone(await self.store.get("nothing")) + + async def test_entry_expires(self): + await self.store.put("name", "taylor", 10) + + self.time.return_value += 9 + self.assertEqual(await self.store.get("name"), "taylor") + + self.time.return_value += 1 + self.assertIs(await self.store.get("name"), MISSING) + self.assertNotIn("name", self.store.storage) + + async def test_forever_never_expires(self): + await self.store.put("name", "taylor", None) + + self.time.return_value += 10**9 + + self.assertEqual(await self.store.get("name"), "taylor") + + async def test_forget_reports_whether_key_existed(self): + await self.store.put("name", "taylor", None) + + self.assertIs(await self.store.forget("name"), True) + self.assertIs(await self.store.forget("name"), False) + self.assertIs(await self.store.get("name"), MISSING) + + async def test_flush_clears_everything(self): + await self.store.put("a", 1, None) + await self.store.put("b", 2, None) + + self.assertIs(await self.store.flush(), True) + self.assertEqual(self.store.storage, {}) diff --git a/fastapi_startkit/tests/cache/test_cache_manager.py b/fastapi_startkit/tests/cache/test_cache_manager.py new file mode 100644 index 00000000..4744b6cd --- /dev/null +++ b/fastapi_startkit/tests/cache/test_cache_manager.py @@ -0,0 +1,80 @@ +from unittest import IsolatedAsyncioTestCase + +from fastapi_startkit.cache import ArrayStore, CacheConfig, CacheManager, Repository +from fastapi_startkit.support.dataclass import Dataclass + + +def make_manager(**overrides): + config = { + "default": "array", + "prefix": "app", + "stores": {"array": {"driver": "array"}, "other": {"driver": "array"}}, + } + config.update(overrides) + return CacheManager(config) + + +class TestCacheManager(IsolatedAsyncioTestCase): + def test_store_resolves_default_and_caches_repository(self): + manager = make_manager() + + repository = manager.store() + + self.assertIsInstance(repository, Repository) + self.assertIsInstance(repository.store, ArrayStore) + self.assertEqual(repository.prefix, "app") + self.assertIs(manager.store("array"), repository) + self.assertIsNot(manager.store("other"), repository) + + async def test_named_stores_are_isolated(self): + manager = make_manager() + + await manager.store("array").put("name", "taylor") + + self.assertIsNone(await manager.store("other").get("name")) + + async def test_unknown_attributes_proxy_to_default_store(self): + manager = make_manager() + + await manager.put("name", "taylor") + + self.assertEqual(await manager.get("name"), "taylor") + self.assertEqual(await manager.store("array").get("name"), "taylor") + self.assertEqual(await manager.remember("name", 60, lambda: "other"), "taylor") + + def test_private_attributes_are_not_proxied(self): + with self.assertRaises(AttributeError): + make_manager()._missing + + def test_undefined_store_raises(self): + with self.assertRaisesRegex(ValueError, r"Cache store \[redis\] is not defined."): + make_manager().store("redis") + + def test_unsupported_driver_raises(self): + manager = make_manager(stores={"file": {"driver": "file"}}, default="file") + + with self.assertRaisesRegex(ValueError, r"Cache driver \[file\] is not supported."): + manager.store() + + async def test_extend_registers_custom_driver(self): + created = [] + + def factory(config): + created.append(config) + return ArrayStore() + + manager = make_manager(stores={"custom": {"driver": "custom", "size": 5}}, default="custom") + + self.assertIs(manager.extend("custom", factory), manager) + await manager.put("name", "taylor") + + self.assertEqual(created, [{"driver": "custom", "size": 5}]) + self.assertEqual(await manager.get("name"), "taylor") + + async def test_accepts_resolved_cache_config(self): + config = {**Dataclass.to_dict(CacheConfig()), "default": "array"} + + manager = CacheManager(config) + await manager.put("name", "taylor") + + self.assertEqual(await manager.store("array").get("name"), "taylor") diff --git a/fastapi_startkit/tests/cache/test_cache_provider.py b/fastapi_startkit/tests/cache/test_cache_provider.py new file mode 100644 index 00000000..79e698d1 --- /dev/null +++ b/fastapi_startkit/tests/cache/test_cache_provider.py @@ -0,0 +1,97 @@ +import os +import tempfile +from dataclasses import dataclass, field +from pathlib import Path +from unittest import IsolatedAsyncioTestCase, mock + +from fastapi_startkit.application import Application +from fastapi_startkit.cache import ArrayStoreConfig, CacheManager, CacheProvider, FileStore +from fastapi_startkit.configuration import config +from fastapi_startkit.facades import Cache + + +@dataclass +class ArrayCacheConfig: + default: str = "array" + prefix: str = "shop" + stores: dict = field(default_factory=lambda: {"array": ArrayStoreConfig(), "second": ArrayStoreConfig()}) + + +class TestCacheProvider(IsolatedAsyncioTestCase): + def setUp(self): + self.enterContext(mock.patch.dict(os.environ)) + for name in ("CACHE_CONNECTION", "CACHE_FILE_PATH"): + os.environ.pop(name, None) + self.base_path = Path(self.enterContext(tempfile.TemporaryDirectory())) + + def make_app(self, provider=(CacheProvider, ArrayCacheConfig)): + return Application(base_path=self.base_path, env="testing", providers=[provider]) + + def test_binds_manager_and_merges_default_config(self): + app = self.make_app(CacheProvider) + + self.assertIsInstance(app.make("cache"), CacheManager) + self.assertEqual(config("cache.default"), "database") + self.assertEqual(config("cache.stores.database.table"), "caches") + + def test_plain_dataclass_config_is_accepted(self): + app = self.make_app() + + self.assertEqual(app.make("cache").config["prefix"], "shop") + self.assertEqual(config("cache.stores.second.driver"), "array") + + async def test_file_store_lives_under_base_path(self): + self.make_app(CacheProvider) + + repository = Cache.store("file") + await repository.put("name", "taylor") + + self.assertIsInstance(repository.store, FileStore) + self.assertEqual(repository.store.directory, self.base_path / "storage/framework/cache/data") + self.assertTrue(repository.store.path("cache:name").is_file()) + self.assertEqual(await Cache.store("file").get("name"), "taylor") + + def test_publishes_config_stub(self): + app = self.make_app() + + self.assertIn("config/cache.py", app.published_resources["cache"].values()) + + async def test_facade_proxies_to_manager(self): + app = self.make_app() + + await Cache.put("name", "taylor", 60) + await Cache.store("second").put("name", "otwell") + + self.assertEqual(await Cache.get("name"), "taylor") + self.assertEqual(await Cache.store("second").get("name"), "otwell") + self.assertIs(await Cache.has("name"), True) + self.assertEqual(await Cache.get("missing", lambda: "fallback"), "fallback") + self.assertIs(Cache.store(), app.make("cache").store()) + + async def test_facade_remember_with_factory(self): + self.make_app() + + async def factory(): + return "computed" + + self.assertEqual(await Cache.remember("value", 60, factory), "computed") + self.assertEqual(await Cache.remember("value", 60, lambda: "other"), "computed") + + async def test_facade_decorator_resolves_the_container_per_call(self): + calls = [] + + @Cache.remember("users", 60) + async def users(role): + calls.append(role) + return [role] + + first = self.make_app() + self.assertEqual(await users("admin"), ["admin"]) + self.assertEqual(await users("admin"), ["admin"]) + self.assertEqual(calls, ["admin"]) + self.assertEqual(users.__name__, "users") + + second = self.make_app() + self.assertEqual(await users("admin"), ["admin"]) + self.assertEqual(calls, ["admin", "admin"]) + self.assertIsNot(first.make("cache"), second.make("cache")) diff --git a/fastapi_startkit/tests/cache/test_config.py b/fastapi_startkit/tests/cache/test_config.py new file mode 100644 index 00000000..cd3e99be --- /dev/null +++ b/fastapi_startkit/tests/cache/test_config.py @@ -0,0 +1,123 @@ +import os +import unittest +from dataclasses import asdict +from unittest import mock + +from fastapi_startkit.cache import ArrayStoreConfig, CacheConfig, DatabaseStoreConfig, FileStoreConfig, RedisStoreConfig + +ENV = ("CACHE_CONNECTION", "CACHE_PREFIX", "CACHE_FILE_PATH", "DB_CACHE_CONNECTION", "DB_CACHE_TABLE") + + +class TestCacheConfig(unittest.TestCase): + def setUp(self): + self.enterContext(mock.patch.dict(os.environ)) + for name in ENV: + os.environ.pop(name, None) + + def test_defaults(self): + config = CacheConfig() + + self.assertEqual(config.default, "database") + self.assertEqual(config.prefix, "cache") + self.assertEqual( + config.stores, + { + "array": ArrayStoreConfig(), + "file": FileStoreConfig(path="storage/framework/cache/data"), + "database": DatabaseStoreConfig(connection=None, table="caches"), + "redis": RedisStoreConfig(connection="cache"), + }, + ) + + def test_reads_environment(self): + os.environ.update( + CACHE_CONNECTION="file", + CACHE_PREFIX="shop", + CACHE_FILE_PATH="var/cache", + DB_CACHE_CONNECTION="sqlite", + DB_CACHE_TABLE="cache_entries", + ) + + config = CacheConfig() + + self.assertEqual(config.default, "file") + self.assertEqual(config.prefix, "shop") + self.assertEqual(config.stores["file"], FileStoreConfig(path="var/cache")) + self.assertEqual(config.stores["database"], DatabaseStoreConfig(connection="sqlite", table="cache_entries")) + + def test_store_drivers(self): + self.assertEqual( + {name: getattr(store, "driver") for name, store in CacheConfig().stores.items()}, + {"array": "array", "file": "file", "database": "database", "redis": "redis"}, + ) + + def test_environment_is_read_per_instance(self): + before = CacheConfig() + os.environ["CACHE_PREFIX"] = "later" + + self.assertEqual(before.prefix, "cache") + self.assertEqual(CacheConfig().prefix, "later") + + def test_empty_environment_value_falls_back_to_default(self): + for name in ("CACHE_CONNECTION", "CACHE_PREFIX"): + with self.subTest(name=name), mock.patch.dict(os.environ, {name: ""}): + config = CacheConfig() + + self.assertEqual((config.default, config.prefix), ("database", "cache")) + + def test_empty_store_environment_values_fall_back_to_defaults(self): + os.environ.update(CACHE_FILE_PATH="", DB_CACHE_TABLE="") + + self.assertEqual(FileStoreConfig().path, "storage/framework/cache/data") + self.assertEqual(DatabaseStoreConfig().table, "caches") + + def test_database_connection_name_is_never_cast(self): + for raw in ("1", "true", "mysql"): + with self.subTest(raw=raw), mock.patch.dict(os.environ, {"DB_CACHE_CONNECTION": raw}): + self.assertEqual(DatabaseStoreConfig().connection, raw) + + def test_explicit_arguments_override_environment(self): + os.environ.update(CACHE_CONNECTION="file", CACHE_PREFIX="shop") + + config = CacheConfig(default="array", prefix="", stores={"array": ArrayStoreConfig()}) + + self.assertEqual(config.default, "array") + self.assertEqual(config.prefix, "") + self.assertEqual(config.stores, {"array": ArrayStoreConfig()}) + + def test_instances_do_not_share_stores(self): + first = CacheConfig() + second = CacheConfig() + + first.stores["custom"] = ArrayStoreConfig() + + self.assertNotIn("custom", second.stores) + self.assertIsNot(first.stores["file"], second.stores["file"]) + + def test_class_attributes_resolve_environment(self): + os.environ.update(CACHE_PREFIX="shop", DB_CACHE_TABLE="cache_entries") + + self.assertEqual(CacheConfig.prefix, "shop") + self.assertEqual(CacheConfig.default, "database") + self.assertEqual(DatabaseStoreConfig.table, "cache_entries") + + def test_unknown_class_attribute_raises(self): + with self.assertRaisesRegex(AttributeError, "has no attribute 'missing'"): + getattr(CacheConfig, "missing") + + def test_converts_to_nested_dict(self): + os.environ["CACHE_FILE_PATH"] = "var/cache" + + self.assertEqual( + asdict(CacheConfig()), + { + "default": "database", + "prefix": "cache", + "stores": { + "array": {"driver": "array"}, + "file": {"driver": "file", "path": "var/cache"}, + "database": {"driver": "database", "connection": None, "table": "caches"}, + "redis": {"driver": "redis", "connection": "cache"}, + }, + }, + ) diff --git a/fastapi_startkit/tests/cache/test_database_store.py b/fastapi_startkit/tests/cache/test_database_store.py new file mode 100644 index 00000000..f99ae6c6 --- /dev/null +++ b/fastapi_startkit/tests/cache/test_database_store.py @@ -0,0 +1,238 @@ +import base64 +import importlib.util +import os +import pickle +import sys +import tempfile +import types +from dataclasses import dataclass +from pathlib import Path +from unittest import IsolatedAsyncioTestCase, mock + +from sqlalchemy.ext.asyncio import create_async_engine +from sqlalchemy.pool import StaticPool + +from fastapi_startkit.application import Application +from fastapi_startkit.cache import MISSING, CacheProvider +from fastapi_startkit.cache.database_store import DatabaseStore +from fastapi_startkit.facades import Cache +from fastapi_startkit.masoniteorm import SQLiteConfig +from fastapi_startkit.masoniteorm.connections.sqlite_connection import SQliteConnection +from fastapi_startkit.masoniteorm.schema import Schema + +MIGRATION = ( + Path(__file__).resolve().parents[2] + / "src/fastapi_startkit/cache/migrations/2026_10_21_000000_create_caches_table.py" +) + + +@dataclass +class User: + name: str + + +class InMemoryDatabase: + def __init__(self): + self.engine = create_async_engine("sqlite+aiosqlite://", poolclass=StaticPool) + self.sqlite = SQliteConnection(self.engine, SQLiteConfig(database=":memory:")) + + def connection(self, name=None): + return self.sqlite + + def get_schema_builder(self): + return Schema(self) + + async def rows(self): + return await self.sqlite.select("SELECT * FROM caches") + + +def load_migration(): + spec = importlib.util.spec_from_file_location("create_caches_table", MIGRATION) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class DatabaseTestCase(IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.database = InMemoryDatabase() + self.migration = load_migration() + + async def asyncTearDown(self): + await self.database.engine.dispose() + + def schema_migration(self): + return self.migration.CreateCachesTable("sqlite", self.database.get_schema_builder()) + + +class TestDatabaseStore(DatabaseTestCase): + async def asyncSetUp(self): + await super().asyncSetUp() + self.time = self.enterContext(mock.patch("time.time", return_value=1_000_000.0)) + self.enterContext(mock.patch.object(self.migration, "config", return_value=None)) + await self.schema_migration().up() + self.store = DatabaseStore(self.database.connection) + + async def insert(self, value): + await ( + self.database.connection() + .query() + .table("caches") + .insert({"key": "name", "value": value, "expiration": None}) + ) + + async def test_missing_key_returns_sentinel(self): + self.assertIs(await self.store.get("name"), MISSING) + + async def test_round_trips_python_objects_as_base64_pickle(self): + self.assertIs(await self.store.put("user", User("taylor"), None), True) + + self.assertEqual(await self.store.get("user"), User("taylor")) + [row] = await self.database.rows() + self.assertEqual(row["key"], "user") + self.assertEqual(pickle.loads(base64.b64decode(row["value"])), User("taylor")) + self.assertIsNone(row["expiration"]) + + async def test_cached_none_is_a_hit(self): + await self.store.put("nothing", None, None) + + self.assertIsNone(await self.store.get("nothing")) + + async def test_put_overwrites_without_duplicating(self): + await self.store.put("name", "taylor", None) + await self.store.put("name", "otwell", 60) + + self.assertEqual(await self.store.get("name"), "otwell") + self.assertEqual(len(await self.database.rows()), 1) + + async def test_put_survives_a_concurrent_insert(self): + original = self.store.query + calls = [] + + def racing_query(): + builder = original() + calls.append(builder) + if len(calls) == 1: + + async def no_rows(values): + await self.insert("") + return 0 + + builder.update = no_rows + return builder + + with mock.patch.object(self.store, "query", racing_query): + self.assertIs(await self.store.put("name", "taylor", None), True) + + self.assertEqual(await self.store.get("name"), "taylor") + self.assertEqual(len(await self.database.rows()), 1) + + async def test_entry_expires_and_row_is_removed(self): + await self.store.put("name", "taylor", 10) + + self.time.return_value += 9 + self.assertEqual(await self.store.get("name"), "taylor") + + self.time.return_value += 1 + self.assertIs(await self.store.get("name"), MISSING) + self.assertEqual(await self.database.rows(), []) + + async def test_forever_never_expires(self): + await self.store.put("name", "taylor", None) + + self.time.return_value += 10**9 + + self.assertEqual(await self.store.get("name"), "taylor") + + async def test_corrupt_value_is_a_miss_and_removed(self): + await self.insert("not base64!") + + self.assertIs(await self.store.get("name"), MISSING) + self.assertEqual(await self.database.rows(), []) + + async def test_value_of_removed_class_is_a_miss_and_removed(self): + for removal in ("module", "class"): + with self.subTest(removal=removal), mock.patch.dict(sys.modules): + module = types.ModuleType("cache_fixture_models") + setattr(module, "Gone", type("Gone", (), {"__module__": "cache_fixture_models"})) + sys.modules["cache_fixture_models"] = module + await self.store.put("gone", getattr(module, "Gone")(), None) + + if removal == "module": + del sys.modules["cache_fixture_models"] + else: + delattr(module, "Gone") + + self.assertIs(await self.store.get("gone"), MISSING) + self.assertEqual(await self.database.rows(), []) + + async def test_expiration_column_is_big_integer(self): + columns = await self.database.sqlite.select("PRAGMA table_info(caches)") + + self.assertEqual({column["name"]: column["type"] for column in columns}["expiration"].upper(), "BIGINT") + + await self.store.put("far", "future", 60 * 60 * 24 * 365 * 100) + [row] = await self.database.rows() + self.assertGreater(row["expiration"], 2**31) + + async def test_forget_reports_whether_key_existed(self): + await self.store.put("name", "taylor", None) + + self.assertIs(await self.store.forget("name"), True) + self.assertIs(await self.store.forget("name"), False) + + async def test_flush_removes_every_entry(self): + await self.store.put("a", 1, None) + await self.store.put("b", 2, None) + + self.assertIs(await self.store.flush(), True) + + self.assertEqual(await self.database.rows(), []) + + async def test_migration_down_drops_table(self): + await self.schema_migration().down() + + self.assertFalse(await self.database.get_schema_builder().has_table("caches")) + + +class TestDatabaseStoreProvider(DatabaseTestCase): + async def asyncSetUp(self): + await super().asyncSetUp() + self.enterContext(mock.patch.dict(os.environ)) + for name in ("CACHE_CONNECTION", "DB_CACHE_CONNECTION", "DB_CACHE_TABLE"): + os.environ.pop(name, None) + self.base_path = Path(self.enterContext(tempfile.TemporaryDirectory())) + + def make_app(self): + app = Application(base_path=self.base_path, env="testing", providers=[CacheProvider]) + app.bind("db", self.database) + return app + + async def test_provider_uses_database_store_by_default(self): + app = self.make_app() + await self.schema_migration().up() + + await Cache.put("name", "taylor") + + self.assertIsInstance(Cache.store().store, DatabaseStore) + self.assertEqual(await Cache.get("name"), "taylor") + [row] = await self.database.rows() + self.assertEqual(row["key"], "cache:name") + self.assertIn( + "databases/migrations/2026_10_21_000000_create_caches_table.py", + app.published_resources["cache"].values(), + ) + + async def test_migration_follows_configured_table(self): + os.environ["DB_CACHE_TABLE"] = "app_cache" + os.environ["CACHE_CONNECTION"] = "database" + self.make_app() + schema = self.database.get_schema_builder() + + await self.schema_migration().up() + await Cache.put("name", "taylor") + + self.assertTrue(await schema.has_table("app_cache")) + self.assertFalse(await schema.has_table("caches")) + self.assertEqual(await Cache.get("name"), "taylor") diff --git a/fastapi_startkit/tests/cache/test_file_store.py b/fastapi_startkit/tests/cache/test_file_store.py new file mode 100644 index 00000000..df1905f2 --- /dev/null +++ b/fastapi_startkit/tests/cache/test_file_store.py @@ -0,0 +1,131 @@ +import hashlib +import pickle +import sys +import tempfile +import types +from dataclasses import dataclass +from pathlib import Path +from unittest import IsolatedAsyncioTestCase, mock + +from fastapi_startkit.cache import MISSING, FileStore + + +@dataclass +class User: + name: str + + +class TestFileStore(IsolatedAsyncioTestCase): + def setUp(self): + self.time = self.enterContext(mock.patch("time.time", return_value=1_000_000.0)) + self.root = Path(self.enterContext(tempfile.TemporaryDirectory())) + self.directory = self.root / "cache" + self.store = FileStore(self.directory) + + def files(self): + return sorted(path.name for path in self.directory.rglob("*") if path.is_file()) + + async def test_missing_key_returns_sentinel(self): + self.assertIs(await self.store.get("name"), MISSING) + + async def test_round_trips_python_objects(self): + self.assertIs(await self.store.put("user", User("taylor"), 60), True) + + self.assertEqual(await self.store.get("user"), User("taylor")) + + async def test_layout_uses_sharded_sha1_paths(self): + await self.store.put("name", "taylor", None) + digest = hashlib.sha1(b"name").hexdigest() + + path = self.directory / digest[:2] / digest[2:4] / digest + self.assertTrue(path.is_file()) + self.assertEqual(pickle.loads(path.read_bytes()), (None, "taylor")) + + async def test_cached_none_is_a_hit(self): + await self.store.put("nothing", None, None) + + self.assertIsNone(await self.store.get("nothing")) + + async def test_entry_expires_and_file_is_removed(self): + await self.store.put("name", "taylor", 10) + + self.time.return_value += 9 + self.assertEqual(await self.store.get("name"), "taylor") + + self.time.return_value += 1 + self.assertIs(await self.store.get("name"), MISSING) + self.assertEqual(self.files(), []) + + async def test_forever_never_expires(self): + await self.store.put("name", "taylor", None) + + self.time.return_value += 10**9 + + self.assertEqual(await self.store.get("name"), "taylor") + + async def test_put_overwrites(self): + await self.store.put("name", "taylor", None) + await self.store.put("name", "otwell", None) + + self.assertEqual(await self.store.get("name"), "otwell") + + async def test_write_is_atomic_and_leaves_no_temporary_file(self): + await self.store.put("name", "taylor", None) + + with mock.patch("os.replace", side_effect=OSError("disk full")): + with self.assertRaisesRegex(OSError, "disk full"): + await self.store.put("name", "otwell", None) + + self.assertEqual(await self.store.get("name"), "taylor") + self.assertEqual(len(self.files()), 1) + + async def test_unpicklable_value_raises_without_writing(self): + with self.assertRaises((pickle.PicklingError, TypeError, AttributeError)): + await self.store.put("lambda", lambda: 1, None) + + self.assertEqual(self.files(), []) + + async def test_corrupt_file_is_a_miss_and_removed(self): + await self.store.put("name", "taylor", None) + self.store.path("name").write_bytes(b"not a pickle") + + self.assertIs(await self.store.get("name"), MISSING) + self.assertEqual(self.files(), []) + + async def test_value_of_removed_class_is_a_miss_and_removed(self): + for removal in ("module", "class"): + with self.subTest(removal=removal), mock.patch.dict(sys.modules): + module = types.ModuleType("cache_fixture_models") + setattr(module, "Gone", type("Gone", (), {"__module__": "cache_fixture_models"})) + sys.modules["cache_fixture_models"] = module + await self.store.put("gone", getattr(module, "Gone")(), None) + + if removal == "module": + del sys.modules["cache_fixture_models"] + else: + delattr(module, "Gone") + + self.assertIs(await self.store.get("gone"), MISSING) + self.assertEqual(self.files(), []) + + async def test_forget_reports_whether_key_existed(self): + await self.store.put("name", "taylor", None) + + self.assertIs(await self.store.forget("name"), True) + self.assertIs(await self.store.forget("name"), False) + self.assertIs(await self.store.get("name"), MISSING) + + async def test_flush_removes_cache_shards_only(self): + await self.store.put("a", 1, None) + await self.store.put("b", 2, None) + (self.directory / ".gitignore").write_text("*\n") + (self.directory / "keep").mkdir() + + self.assertIs(await self.store.flush(), True) + + self.assertIs(await self.store.get("a"), MISSING) + self.assertIs(await self.store.get("b"), MISSING) + self.assertEqual(sorted(path.name for path in self.directory.iterdir()), [".gitignore", "keep"]) + + async def test_flush_without_directory(self): + self.assertIs(await FileStore(self.root / "absent").flush(), True) diff --git a/fastapi_startkit/tests/cache/test_lock.py b/fastapi_startkit/tests/cache/test_lock.py new file mode 100644 index 00000000..1503b8a6 --- /dev/null +++ b/fastapi_startkit/tests/cache/test_lock.py @@ -0,0 +1,258 @@ +import asyncio +import tempfile +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 + + +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_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() + + 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_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() + + 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 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() + + 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] diff --git a/fastapi_startkit/tests/cache/test_memo_store.py b/fastapi_startkit/tests/cache/test_memo_store.py new file mode 100644 index 00000000..7f5df075 --- /dev/null +++ b/fastapi_startkit/tests/cache/test_memo_store.py @@ -0,0 +1,102 @@ +import asyncio +from unittest import IsolatedAsyncioTestCase, mock + +from fastapi_startkit.cache import MISSING, ArrayStore, CacheManager, MemoStore + + +class TestMemoStore(IsolatedAsyncioTestCase): + def setUp(self): + self.inner = ArrayStore() + self.reads = self.enterContext(mock.patch.object(self.inner, "get", wraps=self.inner.get)) + self.store = MemoStore(self.inner) + + async def test_repeated_get_reads_inner_store_once(self): + await self.inner.put("name", "taylor", None) + + self.assertEqual(await self.store.get("name"), "taylor") + self.assertEqual(await self.store.get("name"), "taylor") + self.assertEqual(self.reads.await_count, 1) + + async def test_miss_is_memoized(self): + self.assertIs(await self.store.get("name"), MISSING) + self.assertIs(await self.store.get("name"), MISSING) + self.assertEqual(self.reads.await_count, 1) + + async def test_memoized_value_survives_inner_change(self): + await self.inner.put("name", "taylor", None) + await self.store.get("name") + + await self.inner.put("name", "abigail", None) + + self.assertEqual(await self.store.get("name"), "taylor") + + async def test_put_writes_through_and_invalidates(self): + await self.store.get("name") + + self.assertIs(await self.store.put("name", "taylor", 60), True) + + self.assertEqual(await self.inner.get("name"), "taylor") + self.assertEqual(await self.store.get("name"), "taylor") + self.assertEqual(self.reads.await_count, 3) + + async def test_forget_writes_through_and_invalidates(self): + await self.inner.put("name", "taylor", None) + await self.store.get("name") + + self.assertIs(await self.store.forget("name"), True) + + self.assertIs(await self.store.get("name"), MISSING) + + async def test_flush_clears_memo_and_inner_store(self): + await self.inner.put("name", "taylor", None) + await self.store.get("name") + + self.assertIs(await self.store.flush(), True) + + self.assertEqual(self.inner.storage, {}) + self.assertIs(await self.store.get("name"), MISSING) + + async def test_tasks_do_not_share_memo(self): + await self.inner.put("name", "taylor", None) + await self.store.get("name") + + await asyncio.gather(self.store.get("name"), self.store.get("name")) + + self.assertEqual(self.reads.await_count, 3) + + async def test_stores_keep_separate_memos(self): + other = MemoStore(ArrayStore()) + await self.inner.put("name", "taylor", None) + + await self.store.get("name") + + self.assertIs(await other.get("name"), MISSING) + + +class TestManagerMemo(IsolatedAsyncioTestCase): + def setUp(self): + self.manager = CacheManager( + { + "default": "array", + "prefix": "app", + "stores": {"array": {"driver": "array"}, "other": {"driver": "array"}}, + } + ) + + async def test_memo_wraps_default_store_with_prefix(self): + repository = self.manager.memo() + + self.assertIsInstance(repository.store, MemoStore) + self.assertIs(repository.store.store, self.manager.store().store) + self.assertEqual(repository.prefix, "app") + + async def test_memo_is_cached_per_store(self): + self.assertIs(self.manager.memo(), self.manager.memo("array")) + self.assertIsNot(self.manager.memo(), self.manager.memo("other")) + + async def test_memo_reads_through_to_the_store(self): + await self.manager.put("name", "taylor") + + self.assertEqual(await self.manager.memo().get("name"), "taylor") + self.assertEqual(await self.manager.memo().remember("count", 60, lambda: 1), 1) + self.assertEqual(await self.manager.get("count"), 1) diff --git a/fastapi_startkit/tests/cache/test_redis_store.py b/fastapi_startkit/tests/cache/test_redis_store.py new file mode 100644 index 00000000..c7851634 --- /dev/null +++ b/fastapi_startkit/tests/cache/test_redis_store.py @@ -0,0 +1,199 @@ +import os +import tempfile +from dataclasses import dataclass, field +from pathlib import Path +from unittest import IsolatedAsyncioTestCase, mock + +import fakeredis + +from fastapi_startkit.application import Application +from fastapi_startkit.cache import MISSING, CacheProvider, RedisStoreConfig +from fastapi_startkit.cache.redis_store import RedisStore +from fastapi_startkit.facades import Cache +from fastapi_startkit.redis import RedisManager, RedisProvider + + +@dataclass +class User: + name: str + + +def redis_config(prefix="app_"): + return { + "client": "fake", + "options": {"prefix": prefix}, + "connections": {"default": {"database": 0}, "cache": {"database": 1}}, + } + + +@dataclass +class RedisCacheConfig: + default: str = "redis" + prefix: str = "shop" + stores: dict = field( + default_factory=lambda: {"redis": RedisStoreConfig(), "other": RedisStoreConfig(connection="default")} + ) + + +class RedisStoreTestCase(IsolatedAsyncioTestCase): + prefix = "app_" + + def setUp(self): + self.server = fakeredis.FakeServer() + self.manager = RedisManager(redis_config(self.prefix)).extend("fake", self.client) + self.raw = fakeredis.FakeAsyncRedis(server=self.server, db=1, decode_responses=True) + self.store = RedisStore(lambda: self.manager.connection("cache"), "shop") + + def client(self, parameters): + return fakeredis.FakeAsyncRedis(server=self.server, **parameters) + + +class TestRedisStore(RedisStoreTestCase): + async def test_missing_key_returns_sentinel(self): + self.assertIs(await self.store.get("shop:name"), MISSING) + + async def test_round_trips_python_objects(self): + self.assertIs(await self.store.put("shop:user", User("taylor"), 60), True) + + self.assertEqual(await self.store.get("shop:user"), User("taylor")) + + async def test_cached_none_is_a_hit(self): + await self.store.put("shop:nothing", None, None) + + self.assertIsNone(await self.store.get("shop:nothing")) + + async def test_writes_through_the_connection_prefix_and_database(self): + await self.store.put("shop:name", "taylor", None) + + self.assertEqual(await self.raw.keys("*"), ["app_shop:name"]) + self.assertEqual(await self.raw.ttl("app_shop:name"), -1) + + async def test_ttl_is_set_in_milliseconds(self): + await self.store.put("shop:name", "taylor", 1.5) + + self.assertIn(await self.raw.pttl("app_shop:name"), range(1400, 1501)) + + async def test_sub_millisecond_ttl_rounds_up_to_one_millisecond(self): + connection = self.manager.connection("cache") + with mock.patch.object(connection.client, "set", wraps=connection.client.set) as set_: + await self.store.put("shop:name", "taylor", 0.0001) + + self.assertEqual(set_.call_args.kwargs["px"], 1) + + async def test_corrupt_payload_is_a_miss_and_removed(self): + await self.raw.set("app_shop:name", "not-base64-pickle") + + self.assertIs(await self.store.get("shop:name"), MISSING) + self.assertEqual(await self.raw.exists("app_shop:name"), 0) + + async def test_forget(self): + await self.store.put("shop:name", "taylor", None) + + self.assertIs(await self.store.forget("shop:name"), True) + self.assertIs(await self.store.forget("shop:name"), False) + self.assertIs(await self.store.get("shop:name"), MISSING) + + async def test_flush_removes_only_prefixed_keys(self): + self.store.batch = 2 + for index in range(5): + await self.store.put(f"shop:{index}", index, None) + await self.raw.set("app_shopping", "kept") + await self.raw.set("app_other:1", "kept") + await self.raw.set("shop:1", "kept") + + self.assertIs(await self.store.flush(), True) + + self.assertEqual(sorted(await self.raw.keys("*")), ["app_other:1", "app_shopping", "shop:1"]) + + async def test_flush_never_calls_flushdb(self): + connection = self.manager.connection("cache") + with ( + mock.patch.object(connection.client, "flushdb") as flushdb, + mock.patch.object(connection.client, "flushall") as flushall, + ): + await self.store.flush() + + flushdb.assert_not_called() + flushall.assert_not_called() + + async def test_flush_escapes_glob_characters_in_prefix(self): + store = RedisStore(lambda: self.manager.connection("cache"), "s*[a]") + await store.put("s*[a]:name", "taylor", None) + await self.raw.set("app_sx[a]:name", "kept") + await self.raw.set("app_sa:name", "kept") + + await store.flush() + + self.assertEqual(sorted(await self.raw.keys("*")), ["app_sa:name", "app_sx[a]:name"]) + + def test_empty_prefix_is_rejected(self): + with self.assertRaises(ValueError): + RedisStore(lambda: self.manager.connection("cache"), "") + + +class TestRedisStoreWithoutConnectionPrefix(RedisStoreTestCase): + prefix = "" + + async def test_flush_removes_only_prefixed_keys(self): + await self.store.put("shop:name", "taylor", None) + await self.raw.set("other", "kept") + + await self.store.flush() + + self.assertEqual(await self.raw.keys("*"), ["other"]) + + +class TestRedisCacheProvider(IsolatedAsyncioTestCase): + def setUp(self): + self.enterContext(mock.patch.dict(os.environ)) + os.environ.pop("CACHE_CONNECTION", None) + self.base_path = Path(self.enterContext(tempfile.TemporaryDirectory())) + self.server = fakeredis.FakeServer() + + def make_app(self): + app = Application( + base_path=self.base_path, + env="testing", + providers=[(RedisProvider, lambda: redis_config()), (CacheProvider, RedisCacheConfig)], + ) + app.make("redis").extend("fake", lambda parameters: fakeredis.FakeAsyncRedis(server=self.server, **parameters)) + return app + + def database(self, number): + return fakeredis.FakeAsyncRedis(server=self.server, db=number, decode_responses=True) + + def test_default_store_config_uses_the_cache_connection(self): + self.assertEqual(RedisStoreConfig().connection, "cache") + + async def test_cache_store_redis_uses_the_named_redis_connection(self): + self.make_app() + + await Cache.store("redis").put("name", "taylor", 60) + + self.assertIsInstance(Cache.store("redis").store, RedisStore) + self.assertEqual(await Cache.store("redis").get("name"), "taylor") + self.assertEqual(await self.database(1).keys("*"), ["app_shop:name"]) + + async def test_store_connection_is_selectable(self): + self.make_app() + + await Cache.store("other").put("name", "taylor") + + self.assertEqual(await self.database(0).keys("*"), ["app_shop:name"]) + + async def test_flush_through_facade_keeps_unrelated_keys(self): + self.make_app() + await self.database(1).set("app_queue:1", "job") + + await Cache.put("name", "taylor") + await Cache.flush() + + self.assertEqual(await self.database(1).keys("*"), ["app_queue:1"]) + + async def test_missing_redis_provider_raises_on_first_use(self): + Application(base_path=self.base_path, env="testing", providers=[(CacheProvider, RedisCacheConfig)]) + + repository = Cache.store("redis") + + with self.assertRaisesRegex(RuntimeError, "RedisProvider"): + await repository.get("name") diff --git a/fastapi_startkit/tests/cache/test_remember_decorator.py b/fastapi_startkit/tests/cache/test_remember_decorator.py new file mode 100644 index 00000000..3337d51d --- /dev/null +++ b/fastapi_startkit/tests/cache/test_remember_decorator.py @@ -0,0 +1,128 @@ +import threading +from unittest import IsolatedAsyncioTestCase, mock + +from fastapi_startkit.cache import ArrayStore, CacheManager, Repository + + +class TestRememberDecorator(IsolatedAsyncioTestCase): + def setUp(self): + self.time = self.enterContext(mock.patch("time.time", return_value=1_000_000.0)) + self.store = ArrayStore() + self.cache = Repository(self.store, "app") + + async def test_caches_async_function_result(self): + calls = [] + + @self.cache.remember("users", 60) + async def users(): + calls.append(1) + return ["taylor"] + + self.assertEqual(await users(), ["taylor"]) + self.assertEqual(await users(), ["taylor"]) + self.assertEqual(calls, [1]) + self.assertEqual(list(self.store.storage), ["app:users"]) + + async def test_wraps_sync_function_into_async(self): + @self.cache.remember("total", 60) + def total(): + return 42 + + self.assertEqual(await total(), 42) + self.assertEqual(total.__name__, "total") + + async def test_awaits_awaitable_returned_by_sync_function(self): + async def query(): + return "rows" + + @self.cache.remember("rows", 60) + def rows(): + return query() + + self.assertEqual(await rows(), "rows") + + async def test_each_argument_set_gets_its_own_key(self): + calls = [] + + @self.cache.remember("users", 60) + async def users_by_role(role, active=True): + calls.append((role, active)) + return f"{role}:{active}" + + self.assertEqual(await users_by_role("admin"), "admin:True") + self.assertEqual(await users_by_role("admin"), "admin:True") + self.assertEqual(await users_by_role("editor"), "editor:True") + self.assertEqual(await users_by_role("admin", active=False), "admin:False") + self.assertEqual(await users_by_role("admin", active=False), "admin:False") + + self.assertEqual(calls, [("admin", True), ("editor", True), ("admin", False)]) + self.assertEqual(len(self.store.storage), 3) + self.assertTrue(all(key.startswith("app:users:") for key in self.store.storage)) + + async def test_keyword_order_does_not_change_the_key(self): + calls = [] + + @self.cache.remember("sum", 60) + def add(**numbers): + calls.append(1) + return sum(numbers.values()) + + self.assertEqual(await add(a=1, b=2), 3) + self.assertEqual(await add(b=2, a=1), 3) + self.assertEqual(calls, [1]) + + async def test_unpicklable_arguments_fall_back_to_repr(self): + lock = threading.Lock() + + @self.cache.remember("locked", 60) + def locked(value): + return value is lock + + self.assertIs(await locked(lock), True) + self.assertIs(await locked(lock), True) + + async def test_key_defaults_to_function_path(self): + @self.cache.remember(None, 60) + def answer(): + return 42 + + self.assertEqual(await answer(), 42) + self.assertEqual( + list(self.store.storage), + [f"app:{__name__}.TestRememberDecorator.test_key_defaults_to_function_path..answer"], + ) + + async def test_expires_with_ttl(self): + values = iter(["first", "second"]) + + @self.cache.remember("value", 10) + def value(): + return next(values) + + self.assertEqual(await value(), "first") + self.time.return_value += 10 + self.assertEqual(await value(), "second") + + async def test_shares_entries_with_callable_form(self): + @self.cache.remember("users", 60) + def users(): + return "decorated" + + await self.cache.put("users", "cached") + + self.assertEqual(await users(), "cached") + self.assertEqual(await self.cache.remember("users", 60, lambda: "factory"), "cached") + + def test_factory_form_requires_a_key(self): + with self.assertRaisesRegex(ValueError, "requires a key"): + self.cache.remember(None, 60, lambda: 1) # pyright: ignore[reportCallIssue, reportArgumentType] + + async def test_manager_proxies_decorator_to_default_store(self): + manager = CacheManager({"default": "array", "stores": {"array": {"driver": "array"}}}) + + @manager.remember("name", 60) + def name(): + return "taylor" + + self.assertEqual(await name(), "taylor") + self.assertEqual(await manager.get("name"), "taylor") diff --git a/fastapi_startkit/tests/cache/test_repository.py b/fastapi_startkit/tests/cache/test_repository.py new file mode 100644 index 00000000..ab66cb54 --- /dev/null +++ b/fastapi_startkit/tests/cache/test_repository.py @@ -0,0 +1,127 @@ +from datetime import timedelta +from unittest import IsolatedAsyncioTestCase, mock + +from fastapi_startkit.cache import ArrayStore, Repository + + +class TestRepository(IsolatedAsyncioTestCase): + def setUp(self): + self.time = self.enterContext(mock.patch("time.time", return_value=1_000_000.0)) + self.store = ArrayStore() + self.cache = Repository(self.store, "app") + + async def test_keys_are_prefixed(self): + await self.cache.put("name", "taylor") + + self.assertEqual(list(self.store.storage), ["app:name"]) + self.assertEqual(await self.cache.get("name"), "taylor") + + async def test_no_prefix(self): + await Repository(self.store).put("name", "taylor") + + self.assertEqual(list(self.store.storage), ["name"]) + + async def test_get_returns_default_on_miss(self): + self.assertIsNone(await self.cache.get("missing")) + self.assertEqual(await self.cache.get("missing", "fallback"), "fallback") + + async def test_get_calls_sync_and_async_defaults_only_on_miss(self): + calls = [] + + def sync_default(): + calls.append("sync") + return "sync" + + async def async_default(): + calls.append("async") + return "async" + + self.assertEqual(await self.cache.get("missing", sync_default), "sync") + self.assertEqual(await self.cache.get("missing", async_default), "async") + + await self.cache.put("present", "value") + self.assertEqual(await self.cache.get("present", sync_default), "value") + self.assertEqual(calls, ["sync", "async"]) + + async def test_cached_none_is_not_replaced_by_default(self): + await self.cache.put("nothing", None) + + self.assertIs(await self.cache.has("nothing"), True) + self.assertIsNone(await self.cache.get("nothing", "fallback")) + + async def test_ttl_accepts_seconds_and_timedelta(self): + await self.cache.put("seconds", 1, 60) + await self.cache.put("delta", 2, timedelta(minutes=2)) + + self.time.return_value += 61 + self.assertIs(await self.cache.has("seconds"), False) + self.assertEqual(await self.cache.get("delta"), 2) + + self.time.return_value += 60 + self.assertIs(await self.cache.has("delta"), False) + + async def test_non_positive_ttl_forgets_the_key(self): + await self.cache.put("name", "taylor") + + self.assertIs(await self.cache.put("name", "otwell", 0), False) + self.assertIs(await self.cache.has("name"), False) + self.assertIs(await self.cache.put("name", "otwell", timedelta(seconds=-1)), False) + + async def test_forever(self): + self.assertIs(await self.cache.forever("name", "taylor"), True) + + self.time.return_value += 10**9 + + self.assertEqual(await self.cache.get("name"), "taylor") + + async def test_forget_and_flush(self): + await self.cache.put("a", 1) + await self.cache.put("b", 2) + + self.assertIs(await self.cache.forget("a"), True) + self.assertIs(await self.cache.has("a"), False) + + self.assertIs(await self.cache.flush(), True) + self.assertIs(await self.cache.has("b"), False) + + async def test_remember_runs_factory_once(self): + calls = [] + + def factory(): + calls.append(1) + return ["taylor"] + + self.assertEqual(await self.cache.remember("users", 60, factory), ["taylor"]) + self.assertEqual(await self.cache.remember("users", 60, factory), ["taylor"]) + self.assertEqual(calls, [1]) + + async def test_remember_awaits_async_factory(self): + async def factory(): + return "computed" + + self.assertEqual(await self.cache.remember("value", 60, factory), "computed") + self.assertEqual(await self.cache.get("value"), "computed") + + async def test_remember_awaits_awaitable_returned_by_sync_factory(self): + async def query(): + return "rows" + + self.assertEqual(await self.cache.remember("rows", 60, lambda: query()), "rows") + + async def test_remember_caches_none(self): + calls = [] + + def factory(): + calls.append(1) + + await self.cache.remember("nothing", 60, factory) + await self.cache.remember("nothing", 60, factory) + + self.assertEqual(calls, [1]) + + async def test_remember_recomputes_after_expiry(self): + values = iter(["first", "second"]) + + self.assertEqual(await self.cache.remember("value", 10, lambda: next(values)), "first") + self.time.return_value += 10 + self.assertEqual(await self.cache.remember("value", 10, lambda: next(values)), "second") 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..d2852c3f --- /dev/null +++ b/fastapi_startkit/tests/cache/test_tagged_store.py @@ -0,0 +1,183 @@ +import hashlib +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 + + +class TaggedContract: + async def make_store(self): + raise NotImplementedError + + async def asyncSetUp(self): + self.store = await 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): + 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): + 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 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() + + 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, "") + + 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): + 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")