Skip to content

API Reference

purecache

purecache

__all__ = ['cache', 'LRUCache'] module-attribute

__version__ = '0.1.0' module-attribute

LRUCache

Source code in src/purecache/backends/lru.py
class LRUCache:
    def __init__(self, capacity: int) -> None:
        if capacity < 1:
            raise ValueError(f"capacity must be >= 1, got {capacity}")
        self._capacity = capacity
        self._cache: OrderedDict[str, Any] = OrderedDict()
        self._lock = asyncio.Lock()

    def __len__(self) -> int:
        # No lock — eventually consistent, safe under asyncio's cooperative scheduling
        return len(self._cache)

    async def get(self, key: str) -> Any:
        # Fast path: unlocked read, miss returns immediately without lock overhead
        value = self._cache.get(key, _MISSING)
        if value is _MISSING:
            return None

        async with self._lock:
            try:
                self._cache.move_to_end(key)  # promote to MRU position
            except KeyError:
                # Evicted between unlocked read and lock acquire,
                # it's a valid race, treat as miss
                return None

        return value

    async def put(self, key: str, value: Any) -> None:
        async with self._lock:
            self._cache.pop(key, None)  # remove existing entry if present
            if len(self._cache) >= self._capacity:
                self._cache.popitem(last=False)  # last=False evicts LRU (head)
            self._cache[key] = value  # insert at MRU position (tail)

cache(func, backend, **kwargs)

Source code in src/purecache/decorators.py
def cache(
    func: Callable[..., Any],
    backend: Callable[..., ICacheBackend],
    **kwargs: Any,
):
    cache_backend = backend(**kwargs)

    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        key = generate_key(args, kwargs)
        cached_res = await cache_backend.get(key)
        if cached_res is not None:
            return cached_res

        res = await func(*args, **kwargs)
        await cache_backend.put(key, res)
        return res

    return wrapper