59 lines
1.8 KiB
Python
59 lines
1.8 KiB
Python
"""Simple in-memory Redis replacement for tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import fnmatch
|
|
import time
|
|
from collections.abc import AsyncIterator
|
|
|
|
|
|
class InMemoryRedis:
|
|
"""Subset of redis.asyncio.Redis API backed by an in-memory dict."""
|
|
|
|
def __init__(self) -> None:
|
|
self._store: dict[str, bytes] = {}
|
|
self._expirations: dict[str, float] = {}
|
|
|
|
async def ping(self) -> bool: # pragma: no cover - compatibility shim
|
|
return True
|
|
|
|
async def get(self, name: str) -> bytes | None:
|
|
self._purge_if_expired(name)
|
|
return self._store.get(name)
|
|
|
|
async def set(self, name: str, value: bytes, ex: int | None = None) -> None:
|
|
self._store[name] = value
|
|
if ex is not None:
|
|
self._expirations[name] = time.monotonic() + ex
|
|
elif name in self._expirations:
|
|
self._expirations.pop(name, None)
|
|
|
|
async def delete(self, *names: str) -> int:
|
|
removed = 0
|
|
for name in names:
|
|
if name in self._store:
|
|
del self._store[name]
|
|
removed += 1
|
|
self._expirations.pop(name, None)
|
|
return removed
|
|
|
|
async def close(self) -> None: # pragma: no cover - interface completeness
|
|
self._store.clear()
|
|
self._expirations.clear()
|
|
|
|
async def scan_iter(self, match: str) -> AsyncIterator[str]:
|
|
pattern = match or "*"
|
|
for key in list(self._store.keys()):
|
|
self._purge_if_expired(key)
|
|
for key in self._store.keys():
|
|
if fnmatch.fnmatch(key, pattern):
|
|
yield key
|
|
|
|
def _purge_if_expired(self, name: str) -> None:
|
|
expires_at = self._expirations.get(name)
|
|
if expires_at is None:
|
|
return
|
|
if expires_at <= time.monotonic():
|
|
self._store.pop(name, None)
|
|
self._expirations.pop(name, None)
|