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
42 changes: 30 additions & 12 deletions fastapi_cachex/backends/redis.py
Original file line number Diff line number Diff line change
Expand Up @@ -227,6 +227,29 @@ async def _scan_keys(self, pattern: str) -> list[str]:
if cursor == 0:
return list(keys)

async def _delete_matching(self, pattern: str) -> int:
"""Delete every key matching ``pattern``, one SCAN page at a time.

Each page is deleted as it arrives, so the keyspace is never held in
memory. Deleting keys SCAN already returned is safe: SCAN still returns
every key present for the whole iteration. A key SCAN repeats is
already gone by then, and DEL counts only keys that existed, so it is
not counted twice.

Returns:
How many keys were deleted
"""
cursor = 0
deleted = 0
while True:
cursor, page = await self.client.scan(
cursor, match=pattern, count=_BATCH_SIZE
)
if page:
deleted += await self.client.delete(*page)
if cursor == 0:
return deleted

async def _delete_keys(self, keys: list[str]) -> int:
"""Delete prefixed keys in batches; returns how many existed."""
deleted = 0
Expand Down Expand Up @@ -353,9 +376,7 @@ async def clear(self) -> None:

Only deletes keys within this backend's prefix.
"""
removed = await self._delete_keys(
await self._scan_keys(f"{self._prefix_pattern}*")
)
removed = await self._delete_matching(f"{self._prefix_pattern}*")
logger.debug("Redis CLEAR; removed=%s", removed)

async def clear_path(self, path: str, include_params: bool = False) -> int:
Expand All @@ -377,15 +398,12 @@ async def clear_path(self, path: str, include_params: bool = False) -> int:
f"{self._prefix_pattern}*{CACHE_KEY_SEPARATOR}"
f"{_escape_glob(path)}{CACHE_KEY_SEPARATOR}{suffix}"
)
keys = await self._scan_keys(pattern)
cleared_count = await self._delete_matching(pattern)

# Also match direct keys (custom key formats without separators)
# e.g. key_prefix + "gitlab:template" stored directly via backend.set()
direct_key = self._make_key(path)
if await self.client.exists(direct_key):
keys.append(direct_key)

cleared_count = await self._delete_keys(keys)
# e.g. key_prefix + "gitlab:template" stored directly via backend.set().
# DEL returns 0 for a missing key, so no EXISTS check is needed.
cleared_count += await self.client.delete(self._make_key(path))
logger.debug(
"Redis CLEAR_PATH; path=%s include_params=%s removed=%s",
path,
Expand Down Expand Up @@ -413,14 +431,14 @@ async def clear_pattern(self, pattern: str) -> int:
Number of cache entries cleared
"""
full_pattern = self._prefix_pattern + pattern
cleared_count = await self._delete_keys(await self._scan_keys(full_pattern))
cleared_count = await self._delete_matching(full_pattern)
if (
cleared_count == 0
and self.key_prefix
and pattern.startswith(self.key_prefix)
):
full_pattern = self._prefix_pattern + pattern.removeprefix(self.key_prefix)
cleared_count = await self._delete_keys(await self._scan_keys(full_pattern))
cleared_count = await self._delete_matching(full_pattern)
if cleared_count:
warnings.warn(
f"clear_pattern({pattern!r}) matched only with the backend's "
Expand Down
95 changes: 95 additions & 0 deletions tests/backends/test_redis.py
Original file line number Diff line number Diff line change
Expand Up @@ -854,6 +854,101 @@ async def scan_with_repeats(*args: Any, **kwargs: Any) -> tuple[int, list[str]]:
assert len(keys) == len(set(keys)) == total


def _record_calls(
backend: AsyncRedisCacheBackend, monkeypatch: pytest.MonkeyPatch, *names: str
) -> list[str]:
"""Record, in order, which of the client methods ``names`` are called."""
calls: list[str] = []
for name in names:
real = getattr(backend.client, name)

async def spy(
*args: Any, _real: Any = real, _name: str = name, **kwargs: Any
) -> Any:
calls.append(_name)
return await _real(*args, **kwargs)

monkeypatch.setattr(backend.client, name, spy)
return calls


async def _fill_pages(backend: AsyncRedisCacheBackend) -> int:
total = _BATCH_SIZE * 3
for index in range(total):
await backend.set(
f"GET|||localhost|||/item|||page={index}",
CacheEntry(fingerprint="e", content=b"v"),
)
return total


@requires_redis
@pytest.mark.asyncio
@pytest.mark.parametrize(
"clear",
[
lambda backend: backend.clear(),
lambda backend: backend.clear_pattern("GET|||*"),
lambda backend: backend.clear_path("/item", include_params=True),
],
ids=["clear", "clear_pattern", "clear_path"],
)
async def test_redis_clear_deletes_each_scan_page_as_it_arrives(
async_redis_backend: AsyncRedisCacheBackend,
monkeypatch: pytest.MonkeyPatch,
clear: Any,
) -> None:
"""Keys are deleted page by page, not collected for the whole keyspace (#172)."""
await _fill_pages(async_redis_backend)
calls = _record_calls(async_redis_backend, monkeypatch, "scan", "delete")

await clear(async_redis_backend)

last_scan = len(calls) - 1 - calls[::-1].index("scan")
assert calls.index("delete") < last_scan
assert await async_redis_backend.get_all_keys() == []


@requires_redis
@pytest.mark.asyncio
async def test_redis_clear_path_does_not_check_exists(
async_redis_backend: AsyncRedisCacheBackend,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""DEL already counts only keys that existed, so EXISTS is a wasted trip (#172)."""
await async_redis_backend.set(
"direct:key", CacheEntry(fingerprint="e", content=b"v")
)
calls = _record_calls(async_redis_backend, monkeypatch, "exists")

assert await async_redis_backend.clear_path("direct:key") == 1
assert await async_redis_backend.clear_path("direct:key") == 0
assert calls == []


@requires_redis
@pytest.mark.asyncio
async def test_redis_clear_counts_a_key_scan_repeats_once(
async_redis_backend: AsyncRedisCacheBackend,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A key SCAN returns again on a later page is already deleted (#173, #172)."""
total = await _fill_pages(async_redis_backend)
real_scan = async_redis_backend.client.scan
returned: list[str] = []

async def scan_with_repeats(*args: Any, **kwargs: Any) -> tuple[int, list[str]]:
cursor, page = await real_scan(*args, **kwargs)
page = [*page, *returned[:1]]
returned.extend(page)
return cursor, page

monkeypatch.setattr(async_redis_backend.client, "scan", scan_with_repeats)

assert await async_redis_backend.clear_pattern("GET|||*") == total
assert len(returned) > total # the repeat was actually injected


@requires_redis
@pytest.mark.asyncio
async def test_redis_get_cache_data_reports_absolute_expiry(
Expand Down
Loading