Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions changelog.d/242.added.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
**`MemoryBackend(max_entries=...)` caps the in-memory cache with LRU eviction.**
Without it the cache grows until entries expire, and a client can grow it on
purpose by varying the query string of a cached route. With a limit, storing
an entry past it evicts the least recently used one, expired or not; a read
hit and every write count as a use. The default stays unlimited.
13 changes: 13 additions & 0 deletions docs/BACKENDS.md
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,19 @@ started again there. To stop it on shutdown, `await backend.aclose()` cancels th
task and waits until it has finished (see [Closing a backend](#closing-a-backend)).
`stop_cleanup()` only requests cancellation.

By default the cache has no size limit: entries leave only when they expire, so
memory grows with request rate × `ttl` × body size, per worker, and a client can
grow it on purpose by varying the query string of a cached route.
`MemoryBackend(max_entries=10_000)` caps the number of entries: storing one more
evicts the least recently used entry, expired or not (a read hit and every
write count as a use). Size the limit from the entries you expect times the
workers you run, since each worker has its own cache; the limit counts entries,
not bytes.

```python
backend = MemoryBackend(max_entries=10_000)
```

`clear_pattern()` matches whole keys case-sensitively on every platform, like Redis.
The glob syntax is Python's `fnmatch`, which differs from Redis in two places: negate
a character class with `[!...]` (Redis uses `[^...]`), and escape a special character
Expand Down
70 changes: 63 additions & 7 deletions fastapi_cachex/backends/memory.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import fnmatch
import logging
import time
from collections import OrderedDict
from collections.abc import Callable
from collections.abc import Iterable

Expand Down Expand Up @@ -44,34 +45,86 @@ def _cancel(task: "asyncio.Task[None]") -> None:
loop.call_soon_threadsafe(task.cancel)


def _validate_max_entries(max_entries: int | None) -> int | None:
"""Return ``max_entries`` if it is ``None`` or a positive ``int``.

Raises:
TypeError: If ``max_entries`` is not an ``int`` (``bool`` included)
ValueError: If ``max_entries`` is zero or negative
"""
if max_entries is None:
return None
if isinstance(max_entries, bool) or not isinstance(max_entries, int):
msg = f"max_entries must be an int or None, got {type(max_entries).__name__}"
raise TypeError(msg)
if max_entries <= 0:
msg = f"max_entries must be positive, got {max_entries!r}"
raise ValueError(msg)
return max_entries


class MemoryBackend(BaseCacheBackend):
"""In-memory cache backend implementation.

Manages an in-memory cache dictionary with automatic expiration cleanup.
Cleanup runs in a background task that periodically removes expired entries.
Cleanup is lazily initialized on first cache operation to ensure proper
async context.

Without ``max_entries`` the cache grows until entries expire, by request
rate times TTL times body size, and a client choosing query strings can
grow it on purpose. With ``max_entries`` the cache holds at most that
many entries: storing one more evicts the least recently used entry,
expired or not. A read hit or any write counts as a use.
"""

def __init__(self, cleanup_interval: int = 60) -> None:
def __init__(
self, cleanup_interval: int = 60, *, max_entries: int | None = None
) -> None:
"""Initialize in-memory cache backend.

Args:
cleanup_interval: Interval in seconds between cleanup runs (default: 60)
max_entries: How many entries the cache holds at most (default:
no limit). When full, the least recently used entry is
evicted to make room.

Raises:
ValueError: If ``cleanup_interval`` is not positive
TypeError: If ``max_entries`` is not an ``int`` or ``None``
ValueError: If ``cleanup_interval`` or ``max_entries`` is not
positive
"""
if cleanup_interval <= 0:
# asyncio.sleep() returns at once for these, so the cleanup loop
# would spin, taking the cache lock on every pass.
msg = f"cleanup_interval must be positive, got {cleanup_interval!r}"
raise ValueError(msg)
self.cache: dict[str, CacheItem] = {}
# Ordered from least to most recently used; a plain `dict` for callers
# (`OrderedDict` is one), with `move_to_end` for the LRU bookkeeping.
self.cache: OrderedDict[str, CacheItem] = OrderedDict()
self.lock = asyncio.Lock()
self.cleanup_interval = cleanup_interval
self.max_entries = _validate_max_entries(max_entries)
self._cleanup_task: asyncio.Task[None] | None = None

def _touch(self, key: str) -> None:
"""Mark ``key`` as the most recently used entry. Call under the lock."""
self.cache.move_to_end(key)

def _store(self, key: str, item: CacheItem) -> None:
"""Put ``item`` under ``key`` as the most recently used entry.

Evicts the least recently used entry when that leaves the cache over
``max_entries``. Call under the lock.
"""
self.cache[key] = item
self.cache.move_to_end(key)
if self.max_entries is not None and len(self.cache) > self.max_entries:
evicted, _ = self.cache.popitem(last=False)
logger.debug(
"Memory cache EVICT; key=%s max_entries=%s", evicted, self.max_entries
)

def _ensure_cleanup_started(self) -> None:
"""Ensure a cleanup task runs on the current event loop.

Expand Down Expand Up @@ -144,6 +197,7 @@ async def get(self, key: str) -> CacheEntry | None:
del self.cache[key]
logger.debug("Memory cache EXPIRED; key=%s removed", key)
return None
self._touch(key)
logger.debug("Memory cache HIT; key=%s", key)
return copy.copy(cached_item.value)

Expand All @@ -165,7 +219,7 @@ async def set(self, key: str, value: CacheEntry, ttl: int | None = None) -> None

async with self.lock:
expiry = time.time() + ttl if ttl is not None else None
self.cache[key] = CacheItem(value=copy.copy(value), expiry=expiry)
self._store(key, CacheItem(value=copy.copy(value), expiry=expiry))
logger.debug("Memory cache SET; key=%s ttl=%s", key, ttl)

async def delete(self, key: str) -> bool:
Expand Down Expand Up @@ -211,7 +265,7 @@ async def set_if_absent(
logger.debug("Memory cache SET_IF_ABSENT EXISTS; key=%s", key)
return False
expiry = now + ttl if ttl is not None else None
self.cache[key] = CacheItem(value=copy.copy(value), expiry=expiry)
self._store(key, CacheItem(value=copy.copy(value), expiry=expiry))
logger.debug("Memory cache SET_IF_ABSENT STORED; key=%s ttl=%s", key, ttl)
return True

Expand Down Expand Up @@ -241,6 +295,7 @@ async def expire_if_equals(self, key: str, expected: CacheEntry, ttl: int) -> bo
logger.debug("Memory cache EXPIRE_IF_EQUALS MISMATCH; key=%s", key)
return False
item.expiry = time.time() + ttl
self._touch(key)
logger.debug("Memory cache EXPIRE_IF_EQUALS HIT; key=%s ttl=%s", key, ttl)
return True

Expand All @@ -263,7 +318,7 @@ async def set_if_equals(
logger.debug("Memory cache SET_IF_EQUALS MISMATCH; key=%s", key)
return False
expiry = now + ttl if ttl is not None else None
self.cache[key] = CacheItem(value=copy.copy(value), expiry=expiry)
self._store(key, CacheItem(value=copy.copy(value), expiry=expiry))
logger.debug("Memory cache SET_IF_EQUALS HIT; key=%s ttl=%s", key, ttl)
return True

Expand All @@ -289,10 +344,11 @@ async def increment(self, key: str, delta: int = 1, ttl: int | None = None) -> i
if item is not None and _is_live(item, now):
value = check_counter_range(counter_value(item.value) + delta)
item.value = counter_entry(value)
self._touch(key)
else:
value = delta
expiry = now + ttl if ttl is not None else None
self.cache[key] = CacheItem(value=counter_entry(value), expiry=expiry)
self._store(key, CacheItem(value=counter_entry(value), expiry=expiry))
logger.debug("Memory cache INCREMENT; key=%s value=%s", key, value)
return value

Expand Down
6 changes: 6 additions & 0 deletions i18n/zh-TW/docs/BACKENDS.md
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,12 @@ BackendProxy.set(backend)

清理 task 會在第一次快取呼叫所在的事件迴圈(event loop)上啟動。若之後的呼叫在另一個迴圈上執行(例如第一個迴圈已關閉),task 會在新的迴圈上重新啟動。關閉應用程式時,`await backend.aclose()` 會取消 task 並等待它結束(參見[關閉後端](#closing-a-backend));`stop_cleanup()` 只會要求取消。

預設沒有容量上限:項目只會在過期時離開,因此記憶體用量會隨「請求速率 × `ttl` × 回應大小」成長,而且每個 worker 各自一份;用戶端也可以刻意變換快取路由的查詢字串來撐大它。`MemoryBackend(max_entries=10_000)` 限制項目數量:再多存一筆時,會淘汰最久未使用的項目,不論它是否已過期(讀取命中與每一次寫入都算使用)。請依預期的項目數乘以 worker 數來決定上限,因為每個 worker 各有一份快取;上限計算的是項目數,不是位元組數。

```python
backend = MemoryBackend(max_entries=10_000)
```

`clear_pattern()` 在所有平台上都以區分大小寫的方式比對完整的鍵,與 Redis 相同。萬用字元語法採用 Python 的 `fnmatch`,與 Redis 有兩處不同:否定字元類別要寫 `[!...]`(Redis 為 `[^...]`);跳脫特殊字元要放進中括號,例如 `[*]`(Redis 另外也接受 `\*`)。`*`、`?` 與 `[abc]` 在兩者上的行為相同。

## Redis {#redis}
Expand Down
135 changes: 135 additions & 0 deletions tests/backends/test_memory.py
Original file line number Diff line number Diff line change
Expand Up @@ -1028,3 +1028,138 @@ async def test_mutating_a_written_entry_does_not_change_the_cache() -> None:

for key, n in (("set", 0), ("absent", 1), ("equals", 2)):
assert await backend.get(key) == CacheEntry(fingerprint=f"v{n}", content=b"x")


# --- max_entries: LRU eviction (#242) ---


def _entry(name: str) -> CacheEntry:
return CacheEntry(fingerprint=name, content=name.encode())


@pytest.mark.parametrize("max_entries", [0, -1])
def test_max_entries_must_be_positive(max_entries: int) -> None:
with pytest.raises(ValueError, match="max_entries must be positive"):
MemoryBackend(max_entries=max_entries)


@pytest.mark.parametrize("max_entries", [1.5, True, "3"])
def test_max_entries_must_be_an_int(max_entries: object) -> None:
with pytest.raises(TypeError, match="max_entries must be an int or None"):
MemoryBackend(max_entries=max_entries) # type: ignore[arg-type]


async def test_without_max_entries_nothing_is_evicted() -> None:
backend = MemoryBackend()
assert backend.max_entries is None
for n in range(50):
await backend.set(f"k{n}", _entry(str(n)))

assert len(backend.cache) == 50


async def test_storing_past_max_entries_evicts_the_least_recently_used() -> None:
backend = MemoryBackend(max_entries=2)
await backend.set("a", _entry("a"))
await backend.set("b", _entry("b"))
await backend.set("c", _entry("c"))

assert list(backend.cache) == ["b", "c"]
assert await backend.get("a") is None
assert await backend.delete("a") is False
assert await backend.get("b") == _entry("b")
assert await backend.get("c") == _entry("c")


async def test_a_read_hit_counts_as_a_use() -> None:
backend = MemoryBackend(max_entries=2)
await backend.set("a", _entry("a"))
await backend.set("b", _entry("b"))
assert await backend.get("a") == _entry("a")
await backend.set("c", _entry("c"))

assert list(backend.cache) == ["a", "c"]


async def test_an_expired_entry_is_evicted_like_any_other(clock: Clock) -> None:
"""An expired read is a miss, so it does not keep the entry in place."""
backend = MemoryBackend(max_entries=2)
await backend.set("a", _entry("a"), ttl=1)
await backend.set("b", _entry("b"))
clock.advance(2)
await backend.set("c", _entry("c"))

assert list(backend.cache) == ["b", "c"]


async def test_replacing_a_key_does_not_evict() -> None:
backend = MemoryBackend(max_entries=2)
await backend.set("a", _entry("a"))
await backend.set("b", _entry("b"))
await backend.set("a", _entry("a2"))

assert list(backend.cache) == ["b", "a"]
assert await backend.get("a") == _entry("a2")


async def test_every_write_that_adds_a_key_counts_towards_the_limit() -> None:
backend = MemoryBackend(max_entries=2)
await backend.set("a", _entry("a"))
assert await backend.set_if_absent("b", _entry("b"))
assert list(backend.cache) == ["a", "b"]

assert await backend.increment("counter") == 1
assert list(backend.cache) == ["b", "counter"]

assert await backend.set_if_absent("c", _entry("c"))
assert list(backend.cache) == ["counter", "c"]


async def test_a_write_to_an_existing_key_refreshes_it() -> None:
backend = MemoryBackend(max_entries=2)
await backend.set("counter", counter_entry(1))
await backend.set("a", _entry("a"))
assert await backend.increment("counter") == 2
await backend.set("b", _entry("b"))
assert list(backend.cache) == ["counter", "b"]

assert await backend.set_if_equals("counter", counter_entry(2), _entry("x"))
await backend.set("c", _entry("c"))
assert list(backend.cache) == ["counter", "c"]

assert await backend.expire_if_equals("counter", _entry("x"), ttl=60)
await backend.set("d", _entry("d"))
assert list(backend.cache) == ["counter", "d"]


async def test_a_rejected_conditional_write_is_not_a_use() -> None:
backend = MemoryBackend(max_entries=2)
await backend.set("a", _entry("a"))
await backend.set("b", _entry("b"))
assert not await backend.set_if_absent("a", _entry("a2"))
assert not await backend.set_if_equals("a", _entry("other"), _entry("a2"))
assert not await backend.expire_if_equals("a", _entry("other"), ttl=60)
await backend.set("c", _entry("c"))

assert list(backend.cache) == ["b", "c"]


async def test_monitoring_reads_are_not_a_use() -> None:
backend = MemoryBackend(max_entries=2)
await backend.set("a", _entry("a"))
await backend.set("b", _entry("b"))
assert await backend.get_all_keys() == ["a", "b"]
assert set(await backend.get_cache_data()) == {"a", "b"}
await backend.set("c", _entry("c"))

assert list(backend.cache) == ["b", "c"]


async def test_the_limit_survives_clear() -> None:
backend = MemoryBackend(max_entries=1)
await backend.set("a", _entry("a"))
await backend.clear()
await backend.set("b", _entry("b"))
await backend.set("c", _entry("c"))

assert list(backend.cache) == ["c"]
34 changes: 34 additions & 0 deletions tests/test_cache_max_entries.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
"""`@cache` on a `MemoryBackend(max_entries=...)` stays within the limit (#242)."""

from fastapi import FastAPI
from fastapi.testclient import TestClient

from fastapi_cachex import BackendProxy
from fastapi_cachex import cache
from fastapi_cachex.backends import MemoryBackend


def test_client_chosen_query_strings_cannot_grow_the_cache_past_the_limit() -> None:
backend = MemoryBackend(max_entries=3)
BackendProxy.set(backend)
app = FastAPI()
calls = {"n": 0}

@app.get("/items")
@cache(ttl=3600)
async def items() -> dict[str, int]:
calls["n"] += 1
return {"calls": calls["n"]}

client = TestClient(app)
for junk in range(10):
assert client.get(f"/items?junk={junk}").status_code == 200

assert len(backend.cache) == 3
assert calls["n"] == 10
# The three most recent variants are served from the cache.
assert client.get("/items?junk=9").json() == {"calls": 10}
assert calls["n"] == 10
# The oldest was evicted and renders again.
assert client.get("/items?junk=0").json() == {"calls": 11}
backend.stop_cleanup()
Loading